/*------------------------------------------------------------------------- * * spi.c * Server Programming Interface * * Portions Copyright (c) 1996-2007, PostgreSQL Global Development Group * Portions Copyright (c) 1994, Regents of the University of California * * * IDENTIFICATION * $PostgreSQL: pgsql/src/backend/executor/spi.c,v 1.171 2007/03/13 00:33:40 tgl Exp $ * *------------------------------------------------------------------------- */ #include "postgres.h" #include "access/printtup.h" #include "catalog/heap.h" #include "commands/trigger.h" #include "executor/spi_priv.h" #include "utils/lsyscache.h" #include "utils/memutils.h" #include "utils/typcache.h" uint32 SPI_processed = 0; Oid SPI_lastoid = InvalidOid; SPITupleTable *SPI_tuptable = NULL; int SPI_result; static _SPI_connection *_SPI_stack = NULL; static _SPI_connection *_SPI_current = NULL; static int _SPI_stack_depth = 0; /* allocated size of _SPI_stack */ static int _SPI_connected = -1; static int _SPI_curid = -1; static void _SPI_prepare_plan(const char *src, _SPI_plan *plan); static int _SPI_execute_plan(_SPI_plan *plan, Datum *Values, const char *Nulls, Snapshot snapshot, Snapshot crosscheck_snapshot, bool read_only, long tcount); static int _SPI_pquery(QueryDesc *queryDesc, long tcount); static void _SPI_error_callback(void *arg); static void _SPI_cursor_operation(Portal portal, bool forward, long count, DestReceiver *dest); static _SPI_plan *_SPI_copy_plan(_SPI_plan *plan, int location); static int _SPI_begin_call(bool execmem); static int _SPI_end_call(bool procmem); static MemoryContext _SPI_execmem(void); static MemoryContext _SPI_procmem(void); static bool _SPI_checktuples(void); /* =================== interface functions =================== */ int SPI_connect(void) { int newdepth; /* * When procedure called by Executor _SPI_curid expected to be equal to * _SPI_connected */ if (_SPI_curid != _SPI_connected) return SPI_ERROR_CONNECT; if (_SPI_stack == NULL) { if (_SPI_connected != -1 || _SPI_stack_depth != 0) elog(ERROR, "SPI stack corrupted"); newdepth = 16; _SPI_stack = (_SPI_connection *) MemoryContextAlloc(TopTransactionContext, newdepth * sizeof(_SPI_connection)); _SPI_stack_depth = newdepth; } else { if (_SPI_stack_depth <= 0 || _SPI_stack_depth <= _SPI_connected) elog(ERROR, "SPI stack corrupted"); if (_SPI_stack_depth == _SPI_connected + 1) { newdepth = _SPI_stack_depth * 2; _SPI_stack = (_SPI_connection *) repalloc(_SPI_stack, newdepth * sizeof(_SPI_connection)); _SPI_stack_depth = newdepth; } } /* * We're entering procedure where _SPI_curid == _SPI_connected - 1 */ _SPI_connected++; Assert(_SPI_connected >= 0 && _SPI_connected < _SPI_stack_depth); _SPI_current = &(_SPI_stack[_SPI_connected]); _SPI_current->processed = 0; _SPI_current->lastoid = InvalidOid; _SPI_current->tuptable = NULL; _SPI_current->procCxt = NULL; /* in case we fail to create 'em */ _SPI_current->execCxt = NULL; _SPI_current->connectSubid = GetCurrentSubTransactionId(); /* * Create memory contexts for this procedure * * XXX it would be better to use PortalContext as the parent context, but * we may not be inside a portal (consider deferred-trigger execution). * Perhaps CurTransactionContext would do? For now it doesn't matter * because we clean up explicitly in AtEOSubXact_SPI(). */ _SPI_current->procCxt = AllocSetContextCreate(TopTransactionContext, "SPI Proc", ALLOCSET_DEFAULT_MINSIZE, ALLOCSET_DEFAULT_INITSIZE, ALLOCSET_DEFAULT_MAXSIZE); _SPI_current->execCxt = AllocSetContextCreate(TopTransactionContext, "SPI Exec", ALLOCSET_DEFAULT_MINSIZE, ALLOCSET_DEFAULT_INITSIZE, ALLOCSET_DEFAULT_MAXSIZE); /* ... and switch to procedure's context */ _SPI_current->savedcxt = MemoryContextSwitchTo(_SPI_current->procCxt); return SPI_OK_CONNECT; } int SPI_finish(void) { int res; res = _SPI_begin_call(false); /* live in procedure memory */ if (res < 0) return res; /* Restore memory context as it was before procedure call */ MemoryContextSwitchTo(_SPI_current->savedcxt); /* Release memory used in procedure call */ MemoryContextDelete(_SPI_current->execCxt); _SPI_current->execCxt = NULL; MemoryContextDelete(_SPI_current->procCxt); _SPI_current->procCxt = NULL; /* * Reset result variables, especially SPI_tuptable which is probably * pointing at a just-deleted tuptable */ SPI_processed = 0; SPI_lastoid = InvalidOid; SPI_tuptable = NULL; /* * After _SPI_begin_call _SPI_connected == _SPI_curid. Now we are closing * connection to SPI and returning to upper Executor and so _SPI_connected * must be equal to _SPI_curid. */ _SPI_connected--; _SPI_curid--; if (_SPI_connected == -1) _SPI_current = NULL; else _SPI_current = &(_SPI_stack[_SPI_connected]); return SPI_OK_FINISH; } /* * Clean up SPI state at transaction commit or abort. */ void AtEOXact_SPI(bool isCommit) { /* * Note that memory contexts belonging to SPI stack entries will be freed * automatically, so we can ignore them here. We just need to restore our * static variables to initial state. */ if (isCommit && _SPI_connected != -1) ereport(WARNING, (errcode(ERRCODE_WARNING), errmsg("transaction left non-empty SPI stack"), errhint("Check for missing \"SPI_finish\" calls."))); _SPI_current = _SPI_stack = NULL; _SPI_stack_depth = 0; _SPI_connected = _SPI_curid = -1; SPI_processed = 0; SPI_lastoid = InvalidOid; SPI_tuptable = NULL; } /* * Clean up SPI state at subtransaction commit or abort. * * During commit, there shouldn't be any unclosed entries remaining from * the current subtransaction; we emit a warning if any are found. */ void AtEOSubXact_SPI(bool isCommit, SubTransactionId mySubid) { bool found = false; while (_SPI_connected >= 0) { _SPI_connection *connection = &(_SPI_stack[_SPI_connected]); if (connection->connectSubid != mySubid) break; /* couldn't be any underneath it either */ found = true; /* * Release procedure memory explicitly (see note in SPI_connect) */ if (connection->execCxt) { MemoryContextDelete(connection->execCxt); connection->execCxt = NULL; } if (connection->procCxt) { MemoryContextDelete(connection->procCxt); connection->procCxt = NULL; } /* * Pop the stack entry and reset global variables. Unlike * SPI_finish(), we don't risk switching to memory contexts that might * be already gone. */ _SPI_connected--; _SPI_curid = _SPI_connected; if (_SPI_connected == -1) _SPI_current = NULL; else _SPI_current = &(_SPI_stack[_SPI_connected]); SPI_processed = 0; SPI_lastoid = InvalidOid; SPI_tuptable = NULL; } if (found && isCommit) ereport(WARNING, (errcode(ERRCODE_WARNING), errmsg("subtransaction left non-empty SPI stack"), errhint("Check for missing \"SPI_finish\" calls."))); /* * If we are aborting a subtransaction and there is an open SPI context * surrounding the subxact, clean up to prevent memory leakage. */ if (_SPI_current && !isCommit) { /* free Executor memory the same as _SPI_end_call would do */ MemoryContextResetAndDeleteChildren(_SPI_current->execCxt); /* throw away any partially created tuple-table */ SPI_freetuptable(_SPI_current->tuptable); _SPI_current->tuptable = NULL; } } /* Pushes SPI stack to allow recursive SPI calls */ void SPI_push(void) { _SPI_curid++; } /* Pops SPI stack to allow recursive SPI calls */ void SPI_pop(void) { _SPI_curid--; } /* Restore state of SPI stack after aborting a subtransaction */ void SPI_restore_connection(void) { Assert(_SPI_connected >= 0); _SPI_curid = _SPI_connected - 1; } /* Parse, plan, and execute a query string */ int SPI_execute(const char *src, bool read_only, long tcount) { _SPI_plan plan; int res; if (src == NULL || tcount < 0) return SPI_ERROR_ARGUMENT; res = _SPI_begin_call(true); if (res < 0) return res; plan.plancxt = NULL; /* doesn't have own context */ plan.query = src; plan.nargs = 0; plan.argtypes = NULL; _SPI_prepare_plan(src, &plan); res = _SPI_execute_plan(&plan, NULL, NULL, InvalidSnapshot, InvalidSnapshot, read_only, tcount); _SPI_end_call(true); return res; } /* Obsolete version of SPI_execute */ int SPI_exec(const char *src, long tcount) { return SPI_execute(src, false, tcount); } /* Execute a previously prepared plan */ int SPI_execute_plan(void *plan, Datum *Values, const char *Nulls, bool read_only, long tcount) { int res; if (plan == NULL || tcount < 0) return SPI_ERROR_ARGUMENT; if (((_SPI_plan *) plan)->nargs > 0 && Values == NULL) return SPI_ERROR_PARAM; res = _SPI_begin_call(true); if (res < 0) return res; res = _SPI_execute_plan((_SPI_plan *) plan, Values, Nulls, InvalidSnapshot, InvalidSnapshot, read_only, tcount); _SPI_end_call(true); return res; } /* Obsolete version of SPI_execute_plan */ int SPI_execp(void *plan, Datum *Values, const char *Nulls, long tcount) { return SPI_execute_plan(plan, Values, Nulls, false, tcount); } /* * SPI_execute_snapshot -- identical to SPI_execute_plan, except that we allow * the caller to specify exactly which snapshots to use. This is currently * not documented in spi.sgml because it is only intended for use by RI * triggers. * * Passing snapshot == InvalidSnapshot will select the normal behavior of * fetching a new snapshot for each query. */ int SPI_execute_snapshot(void *plan, Datum *Values, const char *Nulls, Snapshot snapshot, Snapshot crosscheck_snapshot, bool read_only, long tcount) { int res; if (plan == NULL || tcount < 0) return SPI_ERROR_ARGUMENT; if (((_SPI_plan *) plan)->nargs > 0 && Values == NULL) return SPI_ERROR_PARAM; res = _SPI_begin_call(true); if (res < 0) return res; res = _SPI_execute_plan((_SPI_plan *) plan, Values, Nulls, snapshot, crosscheck_snapshot, read_only, tcount); _SPI_end_call(true); return res; } void * SPI_prepare(const char *src, int nargs, Oid *argtypes) { _SPI_plan plan; _SPI_plan *result; if (src == NULL || nargs < 0 || (nargs > 0 && argtypes == NULL)) { SPI_result = SPI_ERROR_ARGUMENT; return NULL; } SPI_result = _SPI_begin_call(true); if (SPI_result < 0) return NULL; plan.plancxt = NULL; /* doesn't have own context */ plan.query = src; plan.nargs = nargs; plan.argtypes = argtypes; _SPI_prepare_plan(src, &plan); /* copy plan to procedure context */ result = _SPI_copy_plan(&plan, _SPI_CPLAN_PROCXT); _SPI_end_call(true); return (void *) result; } void * SPI_saveplan(void *plan) { _SPI_plan *newplan; if (plan == NULL) { SPI_result = SPI_ERROR_ARGUMENT; return NULL; } SPI_result = _SPI_begin_call(false); /* don't change context */ if (SPI_result < 0) return NULL; newplan = _SPI_copy_plan((_SPI_plan *) plan, _SPI_CPLAN_TOPCXT); _SPI_curid--; SPI_result = 0; return (void *) newplan; } int SPI_freeplan(void *plan) { _SPI_plan *spiplan = (_SPI_plan *) plan; if (plan == NULL) return SPI_ERROR_ARGUMENT; MemoryContextDelete(spiplan->plancxt); return 0; } HeapTuple SPI_copytuple(HeapTuple tuple) { MemoryContext oldcxt = NULL; HeapTuple ctuple; if (tuple == NULL) { SPI_result = SPI_ERROR_ARGUMENT; return NULL; } if (_SPI_curid + 1 == _SPI_connected) /* connected */ { if (_SPI_current != &(_SPI_stack[_SPI_curid + 1])) elog(ERROR, "SPI stack corrupted"); oldcxt = MemoryContextSwitchTo(_SPI_current->savedcxt); } ctuple = heap_copytuple(tuple); if (oldcxt) MemoryContextSwitchTo(oldcxt); return ctuple; } HeapTupleHeader SPI_returntuple(HeapTuple tuple, TupleDesc tupdesc) { MemoryContext oldcxt = NULL; HeapTupleHeader dtup; if (tuple == NULL || tupdesc == NULL) { SPI_result = SPI_ERROR_ARGUMENT; return NULL; } /* For RECORD results, make sure a typmod has been assigned */ if (tupdesc->tdtypeid == RECORDOID && tupdesc->tdtypmod < 0) assign_record_type_typmod(tupdesc); if (_SPI_curid + 1 == _SPI_connected) /* connected */ { if (_SPI_current != &(_SPI_stack[_SPI_curid + 1])) elog(ERROR, "SPI stack corrupted"); oldcxt = MemoryContextSwitchTo(_SPI_current->savedcxt); } dtup = (HeapTupleHeader) palloc(tuple->t_len); memcpy((char *) dtup, (char *) tuple->t_data, tuple->t_len); HeapTupleHeaderSetDatumLength(dtup, tuple->t_len); HeapTupleHeaderSetTypeId(dtup, tupdesc->tdtypeid); HeapTupleHeaderSetTypMod(dtup, tupdesc->tdtypmod); if (oldcxt) MemoryContextSwitchTo(oldcxt); return dtup; } HeapTuple SPI_modifytuple(Relation rel, HeapTuple tuple, int natts, int *attnum, Datum *Values, const char *Nulls) { MemoryContext oldcxt = NULL; HeapTuple mtuple; int numberOfAttributes; Datum *v; char *n; int i; if (rel == NULL || tuple == NULL || natts < 0 || attnum == NULL || Values == NULL) { SPI_result = SPI_ERROR_ARGUMENT; return NULL; } if (_SPI_curid + 1 == _SPI_connected) /* connected */ { if (_SPI_current != &(_SPI_stack[_SPI_curid + 1])) elog(ERROR, "SPI stack corrupted"); oldcxt = MemoryContextSwitchTo(_SPI_current->savedcxt); } SPI_result = 0; numberOfAttributes = rel->rd_att->natts; v = (Datum *) palloc(numberOfAttributes * sizeof(Datum)); n = (char *) palloc(numberOfAttributes * sizeof(char)); /* fetch old values and nulls */ heap_deformtuple(tuple, rel->rd_att, v, n); /* replace values and nulls */ for (i = 0; i < natts; i++) { if (attnum[i] <= 0 || attnum[i] > numberOfAttributes) break; v[attnum[i] - 1] = Values[i]; n[attnum[i] - 1] = (Nulls && Nulls[i] == 'n') ? 'n' : ' '; } if (i == natts) /* no errors in *attnum */ { mtuple = heap_formtuple(rel->rd_att, v, n); /* * copy the identification info of the old tuple: t_ctid, t_self, and * OID (if any) */ mtuple->t_data->t_ctid = tuple->t_data->t_ctid; mtuple->t_self = tuple->t_self; mtuple->t_tableOid = tuple->t_tableOid; if (rel->rd_att->tdhasoid) HeapTupleSetOid(mtuple, HeapTupleGetOid(tuple)); } else { mtuple = NULL; SPI_result = SPI_ERROR_NOATTRIBUTE; } pfree(v); pfree(n); if (oldcxt) MemoryContextSwitchTo(oldcxt); return mtuple; } int SPI_fnumber(TupleDesc tupdesc, const char *fname) { int res; Form_pg_attribute sysatt; for (res = 0; res < tupdesc->natts; res++) { if (namestrcmp(&tupdesc->attrs[res]->attname, fname) == 0) return res + 1; } sysatt = SystemAttributeByName(fname, true /* "oid" will be accepted */ ); if (sysatt != NULL) return sysatt->attnum; /* SPI_ERROR_NOATTRIBUTE is different from all sys column numbers */ return SPI_ERROR_NOATTRIBUTE; } char * SPI_fname(TupleDesc tupdesc, int fnumber) { Form_pg_attribute att; SPI_result = 0; if (fnumber > tupdesc->natts || fnumber == 0 || fnumber <= FirstLowInvalidHeapAttributeNumber) { SPI_result = SPI_ERROR_NOATTRIBUTE; return NULL; } if (fnumber > 0) att = tupdesc->attrs[fnumber - 1]; else att = SystemAttributeDefinition(fnumber, true); return pstrdup(NameStr(att->attname)); } char * SPI_getvalue(HeapTuple tuple, TupleDesc tupdesc, int fnumber) { char *result; Datum origval, val; bool isnull; Oid typoid, foutoid; bool typisvarlena; SPI_result = 0; if (fnumber > HeapTupleHeaderGetNatts(tuple->t_data) || fnumber == 0 || fnumber <= FirstLowInvalidHeapAttributeNumber) { SPI_result = SPI_ERROR_NOATTRIBUTE; return NULL; } origval = heap_getattr(tuple, fnumber, tupdesc, &isnull); if (isnull) return NULL; if (fnumber > 0) typoid = tupdesc->attrs[fnumber - 1]->atttypid; else typoid = (SystemAttributeDefinition(fnumber, true))->atttypid; getTypeOutputInfo(typoid, &foutoid, &typisvarlena); /* * If we have a toasted datum, forcibly detoast it here to avoid memory * leakage inside the type's output routine. */ if (typisvarlena) val = PointerGetDatum(PG_DETOAST_DATUM(origval)); else val = origval; result = OidOutputFunctionCall(foutoid, val); /* Clean up detoasted copy, if any */ if (val != origval) pfree(DatumGetPointer(val)); return result; } Datum SPI_getbinval(HeapTuple tuple, TupleDesc tupdesc, int fnumber, bool *isnull) { SPI_result = 0; if (fnumber > HeapTupleHeaderGetNatts(tuple->t_data) || fnumber == 0 || fnumber <= FirstLowInvalidHeapAttributeNumber) { SPI_result = SPI_ERROR_NOATTRIBUTE; *isnull = true; return (Datum) NULL; } return heap_getattr(tuple, fnumber, tupdesc, isnull); } char * SPI_gettype(TupleDesc tupdesc, int fnumber) { Oid typoid; HeapTuple typeTuple; char *result; SPI_result = 0; if (fnumber > tupdesc->natts || fnumber == 0 || fnumber <= FirstLowInvalidHeapAttributeNumber) { SPI_result = SPI_ERROR_NOATTRIBUTE; return NULL; } if (fnumber > 0) typoid = tupdesc->attrs[fnumber - 1]->atttypid; else typoid = (SystemAttributeDefinition(fnumber, true))->atttypid; typeTuple = SearchSysCache(TYPEOID, ObjectIdGetDatum(typoid), 0, 0, 0); if (!HeapTupleIsValid(typeTuple)) { SPI_result = SPI_ERROR_TYPUNKNOWN; return NULL; } result = pstrdup(NameStr(((Form_pg_type) GETSTRUCT(typeTuple))->typname)); ReleaseSysCache(typeTuple); return result; } Oid SPI_gettypeid(TupleDesc tupdesc, int fnumber) { SPI_result = 0; if (fnumber > tupdesc->natts || fnumber == 0 || fnumber <= FirstLowInvalidHeapAttributeNumber) { SPI_result = SPI_ERROR_NOATTRIBUTE; return InvalidOid; } if (fnumber > 0) return tupdesc->attrs[fnumber - 1]->atttypid; else return (SystemAttributeDefinition(fnumber, true))->atttypid; } char * SPI_getrelname(Relation rel) { return pstrdup(RelationGetRelationName(rel)); } char * SPI_getnspname(Relation rel) { return get_namespace_name(RelationGetNamespace(rel)); } void * SPI_palloc(Size size) { MemoryContext oldcxt = NULL; void *pointer; if (_SPI_curid + 1 == _SPI_connected) /* connected */ { if (_SPI_current != &(_SPI_stack[_SPI_curid + 1])) elog(ERROR, "SPI stack corrupted"); oldcxt = MemoryContextSwitchTo(_SPI_current->savedcxt); } pointer = palloc(size); if (oldcxt) MemoryContextSwitchTo(oldcxt); return pointer; } void * SPI_repalloc(void *pointer, Size size) { /* No longer need to worry which context chunk was in... */ return repalloc(pointer, size); } void SPI_pfree(void *pointer) { /* No longer need to worry which context chunk was in... */ pfree(pointer); } void SPI_freetuple(HeapTuple tuple) { /* No longer need to worry which context tuple was in... */ heap_freetuple(tuple); } void SPI_freetuptable(SPITupleTable *tuptable) { if (tuptable != NULL) MemoryContextDelete(tuptable->tuptabcxt); } /* * SPI_cursor_open() * * Open a prepared SPI plan as a portal */ Portal SPI_cursor_open(const char *name, void *plan, Datum *Values, const char *Nulls, bool read_only) { _SPI_plan *spiplan = (_SPI_plan *) plan; List *stmt_list; ParamListInfo paramLI; Snapshot snapshot; MemoryContext oldcontext; Portal portal; int k; /* * Check that the plan is something the Portal code will special-case as * returning one tupleset. */ if (!SPI_is_cursor_plan(spiplan)) { /* try to give a good error message */ Node *stmt; if (list_length(spiplan->stmt_list_list) != 1) ereport(ERROR, (errcode(ERRCODE_INVALID_CURSOR_DEFINITION), errmsg("cannot open multi-query plan as cursor"))); stmt = PortalListGetPrimaryStmt((List *) linitial(spiplan->stmt_list_list)); ereport(ERROR, (errcode(ERRCODE_INVALID_CURSOR_DEFINITION), /* translator: %s is name of a SQL command, eg INSERT */ errmsg("cannot open %s query as cursor", CreateCommandTag(stmt)))); } Assert(list_length(spiplan->stmt_list_list) == 1); stmt_list = (List *) linitial(spiplan->stmt_list_list); /* Reset SPI result (note we deliberately don't touch lastoid) */ SPI_processed = 0; SPI_tuptable = NULL; _SPI_current->processed = 0; _SPI_current->tuptable = NULL; /* Create the portal */ if (name == NULL || name[0] == '\0') { /* Use a random nonconflicting name */ portal = CreateNewPortal(); } else { /* In this path, error if portal of same name already exists */ portal = CreatePortal(name, false, false); } /* Switch to portal's memory and copy the plans to there */ oldcontext = MemoryContextSwitchTo(PortalGetHeapMemory(portal)); stmt_list = copyObject(stmt_list); /* If the plan has parameters, set them up */ if (spiplan->nargs > 0) { /* sizeof(ParamListInfoData) includes the first array element */ paramLI = (ParamListInfo) palloc(sizeof(ParamListInfoData) + (spiplan->nargs - 1) *sizeof(ParamExternData)); paramLI->numParams = spiplan->nargs; for (k = 0; k < spiplan->nargs; k++) { ParamExternData *prm = ¶mLI->params[k]; prm->ptype = spiplan->argtypes[k]; prm->pflags = 0; prm->isnull = (Nulls && Nulls[k] == 'n'); if (prm->isnull) { /* nulls just copy */ prm->value = Values[k]; } else { /* pass-by-ref values must be copied into portal context */ int16 paramTypLen; bool paramTypByVal; get_typlenbyval(prm->ptype, ¶mTypLen, ¶mTypByVal); prm->value = datumCopy(Values[k], paramTypByVal, paramTypLen); } } } else paramLI = NULL; /* * Set up the portal. */ PortalDefineQuery(portal, NULL, /* no statement name */ spiplan->query, CreateCommandTag(PortalListGetPrimaryStmt(stmt_list)), stmt_list, NULL); MemoryContextSwitchTo(oldcontext); /* * Set up options for portal. */ portal->cursorOptions &= ~(CURSOR_OPT_SCROLL | CURSOR_OPT_NO_SCROLL); if (list_length(stmt_list) == 1 && IsA((Node *) linitial(stmt_list), PlannedStmt) && ExecSupportsBackwardScan(((PlannedStmt *) linitial(stmt_list))->planTree)) portal->cursorOptions |= CURSOR_OPT_SCROLL; else portal->cursorOptions |= CURSOR_OPT_NO_SCROLL; /* * Set up the snapshot to use. (PortalStart will do CopySnapshot, so we * skip that here.) */ if (read_only) snapshot = ActiveSnapshot; else { CommandCounterIncrement(); snapshot = GetTransactionSnapshot(); } /* * Start portal execution. */ PortalStart(portal, paramLI, snapshot); Assert(portal->strategy != PORTAL_MULTI_QUERY); /* Return the created portal */ return portal; } /* * SPI_cursor_find() * * Find the portal of an existing open cursor */ Portal SPI_cursor_find(const char *name) { return GetPortalByName(name); } /* * SPI_cursor_fetch() * * Fetch rows in a cursor */ void SPI_cursor_fetch(Portal portal, bool forward, long count) { _SPI_cursor_operation(portal, forward, count, CreateDestReceiver(DestSPI, NULL)); /* we know that the DestSPI receiver doesn't need a destroy call */ } /* * SPI_cursor_move() * * Move in a cursor */ void SPI_cursor_move(Portal portal, bool forward, long count) { _SPI_cursor_operation(portal, forward, count, None_Receiver); } /* * SPI_cursor_close() * * Close a cursor */ void SPI_cursor_close(Portal portal) { if (!PortalIsValid(portal)) elog(ERROR, "invalid portal in SPI cursor operation"); PortalDrop(portal, false); } /* * Returns the Oid representing the type id for argument at argIndex. First * parameter is at index zero. */ Oid SPI_getargtypeid(void *plan, int argIndex) { if (plan == NULL || argIndex < 0 || argIndex >= ((_SPI_plan *) plan)->nargs) { SPI_result = SPI_ERROR_ARGUMENT; return InvalidOid; } return ((_SPI_plan *) plan)->argtypes[argIndex]; } /* * Returns the number of arguments for the prepared plan. */ int SPI_getargcount(void *plan) { if (plan == NULL) { SPI_result = SPI_ERROR_ARGUMENT; return -1; } return ((_SPI_plan *) plan)->nargs; } /* * Returns true if the plan contains exactly one command * and that command returns tuples to the caller (eg, SELECT or * INSERT ... RETURNING, but not SELECT ... INTO). In essence, * the result indicates if the command can be used with SPI_cursor_open * * Parameters * plan: A plan previously prepared using SPI_prepare */ bool SPI_is_cursor_plan(void *plan) { _SPI_plan *spiplan = (_SPI_plan *) plan; if (spiplan == NULL) { SPI_result = SPI_ERROR_ARGUMENT; return false; } if (list_length(spiplan->stmt_list_list) != 1) return false; /* not exactly 1 pre-rewrite command */ switch (ChoosePortalStrategy((List *) linitial(spiplan->stmt_list_list))) { case PORTAL_ONE_SELECT: case PORTAL_ONE_RETURNING: case PORTAL_UTIL_SELECT: /* OK */ return true; case PORTAL_MULTI_QUERY: /* will not return tuples */ break; } return false; } /* * SPI_result_code_string --- convert any SPI return code to a string * * This is often useful in error messages. Most callers will probably * only pass negative (error-case) codes, but for generality we recognize * the success codes too. */ const char * SPI_result_code_string(int code) { static char buf[64]; switch (code) { case SPI_ERROR_CONNECT: return "SPI_ERROR_CONNECT"; case SPI_ERROR_COPY: return "SPI_ERROR_COPY"; case SPI_ERROR_OPUNKNOWN: return "SPI_ERROR_OPUNKNOWN"; case SPI_ERROR_UNCONNECTED: return "SPI_ERROR_UNCONNECTED"; case SPI_ERROR_CURSOR: return "SPI_ERROR_CURSOR"; case SPI_ERROR_ARGUMENT: return "SPI_ERROR_ARGUMENT"; case SPI_ERROR_PARAM: return "SPI_ERROR_PARAM"; case SPI_ERROR_TRANSACTION: return "SPI_ERROR_TRANSACTION"; case SPI_ERROR_NOATTRIBUTE: return "SPI_ERROR_NOATTRIBUTE"; case SPI_ERROR_NOOUTFUNC: return "SPI_ERROR_NOOUTFUNC"; case SPI_ERROR_TYPUNKNOWN: return "SPI_ERROR_TYPUNKNOWN"; case SPI_OK_CONNECT: return "SPI_OK_CONNECT"; case SPI_OK_FINISH: return "SPI_OK_FINISH"; case SPI_OK_FETCH: return "SPI_OK_FETCH"; case SPI_OK_UTILITY: return "SPI_OK_UTILITY"; case SPI_OK_SELECT: return "SPI_OK_SELECT"; case SPI_OK_SELINTO: return "SPI_OK_SELINTO"; case SPI_OK_INSERT: return "SPI_OK_INSERT"; case SPI_OK_DELETE: return "SPI_OK_DELETE"; case SPI_OK_UPDATE: return "SPI_OK_UPDATE"; case SPI_OK_CURSOR: return "SPI_OK_CURSOR"; case SPI_OK_INSERT_RETURNING: return "SPI_OK_INSERT_RETURNING"; case SPI_OK_DELETE_RETURNING: return "SPI_OK_DELETE_RETURNING"; case SPI_OK_UPDATE_RETURNING: return "SPI_OK_UPDATE_RETURNING"; } /* Unrecognized code ... return something useful ... */ sprintf(buf, "Unrecognized SPI code %d", code); return buf; } /* =================== private functions =================== */ /* * spi_dest_startup * Initialize to receive tuples from Executor into SPITupleTable * of current SPI procedure */ void spi_dest_startup(DestReceiver *self, int operation, TupleDesc typeinfo) { SPITupleTable *tuptable; MemoryContext oldcxt; MemoryContext tuptabcxt; /* * When called by Executor _SPI_curid expected to be equal to * _SPI_connected */ if (_SPI_curid != _SPI_connected || _SPI_connected < 0) elog(ERROR, "improper call to spi_dest_startup"); if (_SPI_current != &(_SPI_stack[_SPI_curid])) elog(ERROR, "SPI stack corrupted"); if (_SPI_current->tuptable != NULL) elog(ERROR, "improper call to spi_dest_startup"); oldcxt = _SPI_procmem(); /* switch to procedure memory context */ tuptabcxt = AllocSetContextCreate(CurrentMemoryContext, "SPI TupTable", ALLOCSET_DEFAULT_MINSIZE, ALLOCSET_DEFAULT_INITSIZE, ALLOCSET_DEFAULT_MAXSIZE); MemoryContextSwitchTo(tuptabcxt); _SPI_current->tuptable = tuptable = (SPITupleTable *) palloc(sizeof(SPITupleTable)); tuptable->tuptabcxt = tuptabcxt; tuptable->alloced = tuptable->free = 128; tuptable->vals = (HeapTuple *) palloc(tuptable->alloced * sizeof(HeapTuple)); tuptable->tupdesc = CreateTupleDescCopy(typeinfo); MemoryContextSwitchTo(oldcxt); } /* * spi_printtup * store tuple retrieved by Executor into SPITupleTable * of current SPI procedure */ void spi_printtup(TupleTableSlot *slot, DestReceiver *self) { SPITupleTable *tuptable; MemoryContext oldcxt; /* * When called by Executor _SPI_curid expected to be equal to * _SPI_connected */ if (_SPI_curid != _SPI_connected || _SPI_connected < 0) elog(ERROR, "improper call to spi_printtup"); if (_SPI_current != &(_SPI_stack[_SPI_curid])) elog(ERROR, "SPI stack corrupted"); tuptable = _SPI_current->tuptable; if (tuptable == NULL) elog(ERROR, "improper call to spi_printtup"); oldcxt = MemoryContextSwitchTo(tuptable->tuptabcxt); if (tuptable->free == 0) { tuptable->free = 256; tuptable->alloced += tuptable->free; tuptable->vals = (HeapTuple *) repalloc(tuptable->vals, tuptable->alloced * sizeof(HeapTuple)); } tuptable->vals[tuptable->alloced - tuptable->free] = ExecCopySlotTuple(slot); (tuptable->free)--; MemoryContextSwitchTo(oldcxt); } /* * Static functions */ /* * Parse and plan a querystring. * * At entry, plan->argtypes and plan->nargs must be valid. * * Result lists are stored into *plan. */ static void _SPI_prepare_plan(const char *src, _SPI_plan *plan) { List *raw_parsetree_list; List *stmt_list_list; ListCell *list_item; ErrorContextCallback spierrcontext; Oid *argtypes = plan->argtypes; int nargs = plan->nargs; /* * Increment CommandCounter to see changes made by now. We must do this * to be sure of seeing any schema changes made by a just-preceding SPI * command. (But we don't bother advancing the snapshot, since the * planner generally operates under SnapshotNow rules anyway.) */ CommandCounterIncrement(); /* * Setup error traceback support for ereport() */ spierrcontext.callback = _SPI_error_callback; spierrcontext.arg = (void *) src; spierrcontext.previous = error_context_stack; error_context_stack = &spierrcontext; /* * Parse the request string into a list of raw parse trees. */ raw_parsetree_list = pg_parse_query(src); /* * Do parse analysis and rule rewrite for each raw parsetree. * * We save the results from each raw parsetree as a separate sublist. * This allows _SPI_execute_plan() to know where the boundaries between * original queries fall. */ stmt_list_list = NIL; foreach(list_item, raw_parsetree_list) { Node *parsetree = (Node *) lfirst(list_item); List *query_list; query_list = pg_analyze_and_rewrite(parsetree, src, argtypes, nargs); stmt_list_list = lappend(stmt_list_list, pg_plan_queries(query_list, NULL, false)); } plan->stmt_list_list = stmt_list_list; /* * Pop the error context stack */ error_context_stack = spierrcontext.previous; } /* * Execute the given plan with the given parameter values * * snapshot: query snapshot to use, or InvalidSnapshot for the normal * behavior of taking a new snapshot for each query. * crosscheck_snapshot: for RI use, all others pass InvalidSnapshot * read_only: TRUE for read-only execution (no CommandCounterIncrement) * tcount: execution tuple-count limit, or 0 for none */ static int _SPI_execute_plan(_SPI_plan *plan, Datum *Values, const char *Nulls, Snapshot snapshot, Snapshot crosscheck_snapshot, bool read_only, long tcount) { volatile int my_res = 0; volatile uint32 my_processed = 0; volatile Oid my_lastoid = InvalidOid; SPITupleTable *volatile my_tuptable = NULL; volatile int res = 0; Snapshot saveActiveSnapshot; /* Be sure to restore ActiveSnapshot on error exit */ saveActiveSnapshot = ActiveSnapshot; PG_TRY(); { ListCell *stmt_list_list_item; ErrorContextCallback spierrcontext; int nargs = plan->nargs; ParamListInfo paramLI; /* Convert parameters to form wanted by executor */ if (nargs > 0) { int k; /* sizeof(ParamListInfoData) includes the first array element */ paramLI = (ParamListInfo) palloc(sizeof(ParamListInfoData) + (nargs - 1) *sizeof(ParamExternData)); paramLI->numParams = nargs; for (k = 0; k < nargs; k++) { ParamExternData *prm = ¶mLI->params[k]; prm->value = Values[k]; prm->isnull = (Nulls && Nulls[k] == 'n'); prm->pflags = 0; prm->ptype = plan->argtypes[k]; } } else paramLI = NULL; /* * Setup error traceback support for ereport() */ spierrcontext.callback = _SPI_error_callback; spierrcontext.arg = (void *) plan->query; spierrcontext.previous = error_context_stack; error_context_stack = &spierrcontext; foreach(stmt_list_list_item, plan->stmt_list_list) { List *stmt_list = (List *) lfirst(stmt_list_list_item); ListCell *stmt_list_item; foreach(stmt_list_item, stmt_list) { Node *stmt = (Node *) lfirst(stmt_list_item); bool canSetTag; QueryDesc *qdesc; DestReceiver *dest; _SPI_current->processed = 0; _SPI_current->lastoid = InvalidOid; _SPI_current->tuptable = NULL; if (IsA(stmt, PlannedStmt)) { canSetTag = ((PlannedStmt *) stmt)->canSetTag; } else { /* utilities are canSetTag if only thing in list */ canSetTag = (list_length(stmt_list) == 1); if (IsA(stmt, CopyStmt)) { CopyStmt *cstmt = (CopyStmt *) stmt; if (cstmt->filename == NULL) { my_res = SPI_ERROR_COPY; goto fail; } } else if (IsA(stmt, DeclareCursorStmt) || IsA(stmt, ClosePortalStmt) || IsA(stmt, FetchStmt)) { my_res = SPI_ERROR_CURSOR; goto fail; } else if (IsA(stmt, TransactionStmt)) { my_res = SPI_ERROR_TRANSACTION; goto fail; } } if (read_only && !CommandIsReadOnly(stmt)) ereport(ERROR, (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), /* translator: %s is a SQL statement name */ errmsg("%s is not allowed in a non-volatile function", CreateCommandTag(stmt)))); /* * If not read-only mode, advance the command counter before * each command. */ if (!read_only) CommandCounterIncrement(); dest = CreateDestReceiver(canSetTag ? DestSPI : DestNone, NULL); if (snapshot == InvalidSnapshot) { /* * Default read_only behavior is to use the entry-time * ActiveSnapshot; if read-write, grab a full new snap. */ if (read_only) ActiveSnapshot = CopySnapshot(saveActiveSnapshot); else ActiveSnapshot = CopySnapshot(GetTransactionSnapshot()); } else { /* * We interpret read_only with a specified snapshot to be * exactly that snapshot, but read-write means use the * snap with advancing of command ID. */ ActiveSnapshot = CopySnapshot(snapshot); if (!read_only) ActiveSnapshot->curcid = GetCurrentCommandId(); } if (IsA(stmt, PlannedStmt)) { qdesc = CreateQueryDesc((PlannedStmt *) stmt, ActiveSnapshot, crosscheck_snapshot, dest, paramLI, false); res = _SPI_pquery(qdesc, canSetTag ? tcount : 0); FreeQueryDesc(qdesc); } else { ProcessUtility(stmt, NULL, /* XXX provide query string? */ paramLI, false, /* not top level */ dest, NULL); /* Update "processed" if stmt returned tuples */ if (_SPI_current->tuptable) _SPI_current->processed = _SPI_current->tuptable->alloced - _SPI_current->tuptable->free; res = SPI_OK_UTILITY; } FreeSnapshot(ActiveSnapshot); ActiveSnapshot = NULL; /* * The last canSetTag query sets the status values returned to * the caller. Be careful to free any tuptables not returned, * to avoid intratransaction memory leak. */ if (canSetTag) { my_processed = _SPI_current->processed; my_lastoid = _SPI_current->lastoid; SPI_freetuptable(my_tuptable); my_tuptable = _SPI_current->tuptable; my_res = res; } else { SPI_freetuptable(_SPI_current->tuptable); _SPI_current->tuptable = NULL; } /* we know that the receiver doesn't need a destroy call */ if (res < 0) { my_res = res; goto fail; } } } fail: /* * Pop the error context stack */ error_context_stack = spierrcontext.previous; } PG_CATCH(); { /* Restore global vars and propagate error */ ActiveSnapshot = saveActiveSnapshot; PG_RE_THROW(); } PG_END_TRY(); ActiveSnapshot = saveActiveSnapshot; /* Save results for caller */ SPI_processed = my_processed; SPI_lastoid = my_lastoid; SPI_tuptable = my_tuptable; /* tuptable now is caller's responsibility, not SPI's */ _SPI_current->tuptable = NULL; /* * If none of the queries had canSetTag, we return the last query's result * code, but not its auxiliary results (for backwards compatibility). */ if (my_res == 0) my_res = res; return my_res; } static int _SPI_pquery(QueryDesc *queryDesc, long tcount) { int operation = queryDesc->operation; int res; switch (operation) { case CMD_SELECT: if (queryDesc->plannedstmt->into) /* select into table? */ res = SPI_OK_SELINTO; else if (queryDesc->dest->mydest != DestSPI) { /* Don't return SPI_OK_SELECT if we're discarding result */ res = SPI_OK_UTILITY; } else res = SPI_OK_SELECT; break; case CMD_INSERT: if (queryDesc->plannedstmt->returningLists) res = SPI_OK_INSERT_RETURNING; else res = SPI_OK_INSERT; break; case CMD_DELETE: if (queryDesc->plannedstmt->returningLists) res = SPI_OK_DELETE_RETURNING; else res = SPI_OK_DELETE; break; case CMD_UPDATE: if (queryDesc->plannedstmt->returningLists) res = SPI_OK_UPDATE_RETURNING; else res = SPI_OK_UPDATE; break; default: return SPI_ERROR_OPUNKNOWN; } #ifdef SPI_EXECUTOR_STATS if (ShowExecutorStats) ResetUsage(); #endif AfterTriggerBeginQuery(); ExecutorStart(queryDesc, 0); ExecutorRun(queryDesc, ForwardScanDirection, tcount); _SPI_current->processed = queryDesc->estate->es_processed; _SPI_current->lastoid = queryDesc->estate->es_lastoid; if ((res == SPI_OK_SELECT || queryDesc->plannedstmt->returningLists) && queryDesc->dest->mydest == DestSPI) { if (_SPI_checktuples()) elog(ERROR, "consistency check on SPI tuple count failed"); } /* Take care of any queued AFTER triggers */ AfterTriggerEndQuery(queryDesc->estate); ExecutorEnd(queryDesc); #ifdef SPI_EXECUTOR_STATS if (ShowExecutorStats) ShowUsage("SPI EXECUTOR STATS"); #endif return res; } /* * _SPI_error_callback * * Add context information when a query invoked via SPI fails */ static void _SPI_error_callback(void *arg) { const char *query = (const char *) arg; int syntaxerrposition; /* * If there is a syntax error position, convert to internal syntax error; * otherwise treat the query as an item of context stack */ syntaxerrposition = geterrposition(); if (syntaxerrposition > 0) { errposition(0); internalerrposition(syntaxerrposition); internalerrquery(query); } else errcontext("SQL statement \"%s\"", query); } /* * _SPI_cursor_operation() * * Do a FETCH or MOVE in a cursor */ static void _SPI_cursor_operation(Portal portal, bool forward, long count, DestReceiver *dest) { long nfetched; /* Check that the portal is valid */ if (!PortalIsValid(portal)) elog(ERROR, "invalid portal in SPI cursor operation"); /* Push the SPI stack */ if (_SPI_begin_call(true) < 0) elog(ERROR, "SPI cursor operation called while not connected"); /* Reset the SPI result (note we deliberately don't touch lastoid) */ SPI_processed = 0; SPI_tuptable = NULL; _SPI_current->processed = 0; _SPI_current->tuptable = NULL; /* Run the cursor */ nfetched = PortalRunFetch(portal, forward ? FETCH_FORWARD : FETCH_BACKWARD, count, dest); /* * Think not to combine this store with the preceding function call. If * the portal contains calls to functions that use SPI, then SPI_stack is * likely to move around while the portal runs. When control returns, * _SPI_current will point to the correct stack entry... but the pointer * may be different than it was beforehand. So we must be sure to re-fetch * the pointer after the function call completes. */ _SPI_current->processed = nfetched; if (dest->mydest == DestSPI && _SPI_checktuples()) elog(ERROR, "consistency check on SPI tuple count failed"); /* Put the result into place for access by caller */ SPI_processed = _SPI_current->processed; SPI_tuptable = _SPI_current->tuptable; /* tuptable now is caller's responsibility, not SPI's */ _SPI_current->tuptable = NULL; /* Pop the SPI stack */ _SPI_end_call(true); } static MemoryContext _SPI_execmem(void) { return MemoryContextSwitchTo(_SPI_current->execCxt); } static MemoryContext _SPI_procmem(void) { return MemoryContextSwitchTo(_SPI_current->procCxt); } /* * _SPI_begin_call: begin a SPI operation within a connected procedure */ static int _SPI_begin_call(bool execmem) { if (_SPI_curid + 1 != _SPI_connected) return SPI_ERROR_UNCONNECTED; _SPI_curid++; if (_SPI_current != &(_SPI_stack[_SPI_curid])) elog(ERROR, "SPI stack corrupted"); if (execmem) /* switch to the Executor memory context */ _SPI_execmem(); return 0; } /* * _SPI_end_call: end a SPI operation within a connected procedure * * Note: this currently has no failure return cases, so callers don't check */ static int _SPI_end_call(bool procmem) { /* * We're returning to procedure where _SPI_curid == _SPI_connected - 1 */ _SPI_curid--; if (procmem) /* switch to the procedure memory context */ { _SPI_procmem(); /* and free Executor memory */ MemoryContextResetAndDeleteChildren(_SPI_current->execCxt); } return 0; } static bool _SPI_checktuples(void) { uint32 processed = _SPI_current->processed; SPITupleTable *tuptable = _SPI_current->tuptable; bool failed = false; if (tuptable == NULL) /* spi_dest_startup was not called */ failed = true; else if (processed != (tuptable->alloced - tuptable->free)) failed = true; return failed; } static _SPI_plan * _SPI_copy_plan(_SPI_plan *plan, int location) { _SPI_plan *newplan; MemoryContext oldcxt; MemoryContext plancxt; MemoryContext parentcxt; /* Determine correct parent for the plan's memory context */ if (location == _SPI_CPLAN_PROCXT) parentcxt = _SPI_current->procCxt; else if (location == _SPI_CPLAN_TOPCXT) parentcxt = TopMemoryContext; else /* (this case not currently used) */ parentcxt = CurrentMemoryContext; /* * Create a memory context for the plan. We don't expect the plan to be * very large, so use smaller-than-default alloc parameters. */ plancxt = AllocSetContextCreate(parentcxt, "SPI Plan", ALLOCSET_SMALL_MINSIZE, ALLOCSET_SMALL_INITSIZE, ALLOCSET_SMALL_MAXSIZE); oldcxt = MemoryContextSwitchTo(plancxt); /* Copy the SPI plan into its own context */ newplan = (_SPI_plan *) palloc(sizeof(_SPI_plan)); newplan->plancxt = plancxt; newplan->query = pstrdup(plan->query); newplan->stmt_list_list = (List *) copyObject(plan->stmt_list_list); newplan->nargs = plan->nargs; if (plan->nargs > 0) { newplan->argtypes = (Oid *) palloc(plan->nargs * sizeof(Oid)); memcpy(newplan->argtypes, plan->argtypes, plan->nargs * sizeof(Oid)); } else newplan->argtypes = NULL; MemoryContextSwitchTo(oldcxt); return newplan; }