mirrors/postgres
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity
postgres/src/backend/executor/spi.c
Dean Rasheed c649fa24a4 Add RETURNING support to MERGE. 
This allows a RETURNING clause to be appended to a MERGE query, to
return values based on each row inserted, updated, or deleted. As with
plain INSERT, UPDATE, and DELETE commands, the returned values are
based on the new contents of the target table for INSERT and UPDATE
actions, and on its old contents for DELETE actions. Values from the
source relation may also be returned.

As with INSERT/UPDATE/DELETE, the output of MERGE ... RETURNING may be
used as the source relation for other operations such as WITH queries
and COPY commands.

Additionally, a special function merge_action() is provided, which
returns 'INSERT', 'UPDATE', or 'DELETE', depending on the action
executed for each row. The merge_action() function can be used
anywhere in the RETURNING list, including in arbitrary expressions and
subqueries, but it is an error to use it anywhere outside of a MERGE
query's RETURNING list.

Dean Rasheed, reviewed by Isaac Morland, Vik Fearing, Alvaro Herrera,
Gurjeet Singh, Jian He, Jeff Davis, Merlin Moncure, Peter Eisentraut,
and Wolfgang Walther.

Discussion: http://postgr.es/m/CAEZATCWePEGQR5LBn-vD6SfeLZafzEm2Qy_L_Oky2=qw2w3Pzg@mail.gmail.com
2024-03-17 13:58:59 +00:00

3385 lines
84 KiB
C
Raw Blame History

/*-------------------------------------------------------------------------
*
* spi.c
* Server Programming Interface
*
* Portions Copyright (c) 1996-2024, PostgreSQL Global Development Group
* Portions Copyright (c) 1994, Regents of the University of California
*
*
* IDENTIFICATION
* src/backend/executor/spi.c
*
*-------------------------------------------------------------------------
*/
#include "postgres.h"
#include "access/htup_details.h"
#include "access/printtup.h"
#include "access/sysattr.h"
#include "access/xact.h"
#include "catalog/heap.h"
#include "catalog/pg_type.h"
#include "commands/trigger.h"
#include "executor/executor.h"
#include "executor/spi_priv.h"
#include "tcop/pquery.h"
#include "tcop/utility.h"
#include "utils/builtins.h"
#include "utils/datum.h"
#include "utils/lsyscache.h"
#include "utils/memutils.h"
#include "utils/rel.h"
#include "utils/snapmgr.h"
#include "utils/syscache.h"
#include "utils/typcache.h"
/*
* These global variables are part of the API for various SPI functions
* (a horrible API choice, but it's too late now). To reduce the risk of
* interference between different SPI callers, we save and restore them
* when entering/exiting a SPI nesting level.
*/
uint64 SPI_processed = 0;
SPITupleTable *SPI_tuptable = NULL;
int SPI_result = 0;
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; /* current stack index */
typedef struct SPICallbackArg
{
const char *query;
RawParseMode mode;
} SPICallbackArg;
static Portal SPI_cursor_open_internal(const char *name, SPIPlanPtr plan,
ParamListInfo paramLI, bool read_only);
static void _SPI_prepare_plan(const char *src, SPIPlanPtr plan);
static void _SPI_prepare_oneshot_plan(const char *src, SPIPlanPtr plan);
static int _SPI_execute_plan(SPIPlanPtr plan, const SPIExecuteOptions *options,
Snapshot snapshot, Snapshot crosscheck_snapshot,
bool fire_triggers);
static ParamListInfo _SPI_convert_params(int nargs, Oid *argtypes,
Datum *Values, const char *Nulls);
static int _SPI_pquery(QueryDesc *queryDesc, bool fire_triggers, uint64 tcount);
static void _SPI_error_callback(void *arg);
static void _SPI_cursor_operation(Portal portal,
FetchDirection direction, long count,
DestReceiver *dest);
static SPIPlanPtr _SPI_make_plan_non_temp(SPIPlanPtr plan);
static SPIPlanPtr _SPI_save_plan(SPIPlanPtr plan);
static int _SPI_begin_call(bool use_exec);
static int _SPI_end_call(bool use_exec);
static MemoryContext _SPI_execmem(void);
static MemoryContext _SPI_procmem(void);
static bool _SPI_checktuples(void);
/* =================== interface functions =================== */
int
SPI_connect(void)
{
return SPI_connect_ext(0);
}
int
SPI_connect_ext(int options)
{
int newdepth;
/* Enlarge stack if necessary */
if (_SPI_stack == NULL)
{
if (_SPI_connected != -1 || _SPI_stack_depth != 0)
elog(ERROR, "SPI stack corrupted");
newdepth = 16;
_SPI_stack = (_SPI_connection *)
MemoryContextAlloc(TopMemoryContext,
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;
}
}
/* Enter new stack level */
_SPI_connected++;
Assert(_SPI_connected >= 0 && _SPI_connected < _SPI_stack_depth);
_SPI_current = &(_SPI_stack[_SPI_connected]);
_SPI_current->processed = 0;
_SPI_current->tuptable = NULL;
_SPI_current->execSubid = InvalidSubTransactionId;
slist_init(&_SPI_current->tuptables);
_SPI_current->procCxt = NULL; /* in case we fail to create 'em */
_SPI_current->execCxt = NULL;
_SPI_current->connectSubid = GetCurrentSubTransactionId();
_SPI_current->queryEnv = NULL;
_SPI_current->atomic = (options & SPI_OPT_NONATOMIC ? false : true);
_SPI_current->internal_xact = false;
_SPI_current->outer_processed = SPI_processed;
_SPI_current->outer_tuptable = SPI_tuptable;
_SPI_current->outer_result = SPI_result;
/*
* Create memory contexts for this procedure
*
* In atomic contexts (the normal case), we use TopTransactionContext,
* otherwise PortalContext, so that it lives across transaction
* boundaries.
*
* XXX It could be better to use PortalContext as the parent context in
* all cases, but we may not be inside a portal (consider deferred-trigger
* execution). Perhaps CurTransactionContext could be an option? For now
* it doesn't matter because we clean up explicitly in AtEOSubXact_SPI();
* but see also AtEOXact_SPI().
*/
_SPI_current->procCxt = AllocSetContextCreate(_SPI_current->atomic ? TopTransactionContext : PortalContext,
"SPI Proc",
ALLOCSET_DEFAULT_SIZES);
_SPI_current->execCxt = AllocSetContextCreate(_SPI_current->atomic ? TopTransactionContext : _SPI_current->procCxt,
"SPI Exec",
ALLOCSET_DEFAULT_SIZES);
/* ... and switch to procedure's context */
_SPI_current->savedcxt = MemoryContextSwitchTo(_SPI_current->procCxt);
/*
* Reset API global variables so that current caller cannot accidentally
* depend on state of an outer caller.
*/
SPI_processed = 0;
SPI_tuptable = NULL;
SPI_result = 0;
return SPI_OK_CONNECT;
}
int
SPI_finish(void)
{
int res;
res = _SPI_begin_call(false); /* just check we're connected */
if (res < 0)
return res;
/* Restore memory context as it was before procedure call */
MemoryContextSwitchTo(_SPI_current->savedcxt);
/* Release memory used in procedure call (including tuptables) */
MemoryContextDelete(_SPI_current->execCxt);
_SPI_current->execCxt = NULL;
MemoryContextDelete(_SPI_current->procCxt);
_SPI_current->procCxt = NULL;
/*
* Restore outer API variables, especially SPI_tuptable which is probably
* pointing at a just-deleted tuptable
*/
SPI_processed = _SPI_current->outer_processed;
SPI_tuptable = _SPI_current->outer_tuptable;
SPI_result = _SPI_current->outer_result;
/* Exit stack level */
_SPI_connected--;
if (_SPI_connected < 0)
_SPI_current = NULL;
else
_SPI_current = &(_SPI_stack[_SPI_connected]);
return SPI_OK_FINISH;
}
/*
* SPI_start_transaction is a no-op, kept for backwards compatibility.
* SPI callers are *always* inside a transaction.
*/
void
SPI_start_transaction(void)
{
}
static void
_SPI_commit(bool chain)
{
MemoryContext oldcontext = CurrentMemoryContext;
SavedTransactionCharacteristics savetc;
/*
* Complain if we are in a context that doesn't permit transaction
* termination. (Note: here and _SPI_rollback should be the only places
* that throw ERRCODE_INVALID_TRANSACTION_TERMINATION, so that callers can
* test for that with security that they know what happened.)
*/
if (_SPI_current->atomic)
ereport(ERROR,
(errcode(ERRCODE_INVALID_TRANSACTION_TERMINATION),
errmsg("invalid transaction termination")));
/*
* This restriction is required by PLs implemented on top of SPI. They
* use subtransactions to establish exception blocks that are supposed to
* be rolled back together if there is an error. Terminating the
* top-level transaction in such a block violates that idea. A future PL
* implementation might have different ideas about this, in which case
* this restriction would have to be refined or the check possibly be
* moved out of SPI into the PLs. Note however that the code below relies
* on not being within a subtransaction.
*/
if (IsSubTransaction())
ereport(ERROR,
(errcode(ERRCODE_INVALID_TRANSACTION_TERMINATION),
errmsg("cannot commit while a subtransaction is active")));
if (chain)
SaveTransactionCharacteristics(&savetc);
/* Catch any error occurring during the COMMIT */
PG_TRY();
{
/* Protect current SPI stack entry against deletion */
_SPI_current->internal_xact = true;
/*
* Hold any pinned portals that any PLs might be using. We have to do
* this before changing transaction state, since this will run
* user-defined code that might throw an error.
*/
HoldPinnedPortals();
/* Release snapshots associated with portals */
ForgetPortalSnapshots();
/* Do the deed */
CommitTransactionCommand();
/* Immediately start a new transaction */
StartTransactionCommand();
if (chain)
RestoreTransactionCharacteristics(&savetc);
MemoryContextSwitchTo(oldcontext);
_SPI_current->internal_xact = false;
}
PG_CATCH();
{
ErrorData *edata;
/* Save error info in caller's context */
MemoryContextSwitchTo(oldcontext);
edata = CopyErrorData();
FlushErrorState();
/*
* Abort the failed transaction. If this fails too, we'll just
* propagate the error out ... there's not that much we can do.
*/
AbortCurrentTransaction();
/* ... and start a new one */
StartTransactionCommand();
if (chain)
RestoreTransactionCharacteristics(&savetc);
MemoryContextSwitchTo(oldcontext);
_SPI_current->internal_xact = false;
/* Now that we've cleaned up the transaction, re-throw the error */
ReThrowError(edata);
}
PG_END_TRY();
}
void
SPI_commit(void)
{
_SPI_commit(false);
}
void
SPI_commit_and_chain(void)
{
_SPI_commit(true);
}
static void
_SPI_rollback(bool chain)
{
MemoryContext oldcontext = CurrentMemoryContext;
SavedTransactionCharacteristics savetc;
/* see under SPI_commit() */
if (_SPI_current->atomic)
ereport(ERROR,
(errcode(ERRCODE_INVALID_TRANSACTION_TERMINATION),
errmsg("invalid transaction termination")));
/* see under SPI_commit() */
if (IsSubTransaction())
ereport(ERROR,
(errcode(ERRCODE_INVALID_TRANSACTION_TERMINATION),
errmsg("cannot roll back while a subtransaction is active")));
if (chain)
SaveTransactionCharacteristics(&savetc);
/* Catch any error occurring during the ROLLBACK */
PG_TRY();
{
/* Protect current SPI stack entry against deletion */
_SPI_current->internal_xact = true;
/*
* Hold any pinned portals that any PLs might be using. We have to do
* this before changing transaction state, since this will run
* user-defined code that might throw an error, and in any case
* couldn't be run in an already-aborted transaction.
*/
HoldPinnedPortals();
/* Release snapshots associated with portals */
ForgetPortalSnapshots();
/* Do the deed */
AbortCurrentTransaction();
/* Immediately start a new transaction */
StartTransactionCommand();
if (chain)
RestoreTransactionCharacteristics(&savetc);
MemoryContextSwitchTo(oldcontext);
_SPI_current->internal_xact = false;
}
PG_CATCH();
{
ErrorData *edata;
/* Save error info in caller's context */
MemoryContextSwitchTo(oldcontext);
edata = CopyErrorData();
FlushErrorState();
/*
* Try again to abort the failed transaction. If this fails too,
* we'll just propagate the error out ... there's not that much we can
* do.
*/
AbortCurrentTransaction();
/* ... and start a new one */
StartTransactionCommand();
if (chain)
RestoreTransactionCharacteristics(&savetc);
MemoryContextSwitchTo(oldcontext);
_SPI_current->internal_xact = false;
/* Now that we've cleaned up the transaction, re-throw the error */
ReThrowError(edata);
}
PG_END_TRY();
}
void
SPI_rollback(void)
{
_SPI_rollback(false);
}
void
SPI_rollback_and_chain(void)
{
_SPI_rollback(true);
}
/*
* Clean up SPI state at transaction commit or abort.
*/
void
AtEOXact_SPI(bool isCommit)
{
bool found = false;
/*
* Pop stack entries, stopping if we find one marked internal_xact (that
* one belongs to the caller of SPI_commit or SPI_rollback).
*/
while (_SPI_connected >= 0)
{
_SPI_connection *connection = &(_SPI_stack[_SPI_connected]);
if (connection->internal_xact)
break;
found = true;
/*
* We need not release the procedure's memory contexts explicitly, as
* they'll go away automatically when their parent context does; see
* notes in SPI_connect_ext.
*/
/*
* Restore outer global variables and pop the stack entry. Unlike
* SPI_finish(), we don't risk switching to memory contexts that might
* be already gone.
*/
SPI_processed = connection->outer_processed;
SPI_tuptable = connection->outer_tuptable;
SPI_result = connection->outer_result;
_SPI_connected--;
if (_SPI_connected < 0)
_SPI_current = NULL;
else
_SPI_current = &(_SPI_stack[_SPI_connected]);
}
/* We should only find entries to pop during an ABORT. */
if (found && isCommit)
ereport(WARNING,
(errcode(ERRCODE_WARNING),
errmsg("transaction left non-empty SPI stack"),
errhint("Check for missing \"SPI_finish\" calls.")));
}
/*
* 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 */
if (connection->internal_xact)
break;
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;
}
/*
* Restore outer global variables and pop the stack entry. Unlike
* SPI_finish(), we don't risk switching to memory contexts that might
* be already gone.
*/
SPI_processed = connection->outer_processed;
SPI_tuptable = connection->outer_tuptable;
SPI_result = connection->outer_result;
_SPI_connected--;
if (_SPI_connected < 0)
_SPI_current = NULL;
else
_SPI_current = &(_SPI_stack[_SPI_connected]);
}
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)
{
slist_mutable_iter siter;
/*
* Throw away executor state if current executor operation was started
* within current subxact (essentially, force a _SPI_end_call(true)).
*/
if (_SPI_current->execSubid >= mySubid)
{
_SPI_current->execSubid = InvalidSubTransactionId;
MemoryContextReset(_SPI_current->execCxt);
}
/* throw away any tuple tables created within current subxact */
slist_foreach_modify(siter, &_SPI_current->tuptables)
{
SPITupleTable *tuptable;
tuptable = slist_container(SPITupleTable, next, siter.cur);
if (tuptable->subid >= mySubid)
{
/*
* If we used SPI_freetuptable() here, its internal search of
* the tuptables list would make this operation O(N^2).
* Instead, just free the tuptable manually. This should
* match what SPI_freetuptable() does.
*/
slist_delete_current(&siter);
if (tuptable == _SPI_current->tuptable)
_SPI_current->tuptable = NULL;
if (tuptable == SPI_tuptable)
SPI_tuptable = NULL;
MemoryContextDelete(tuptable->tuptabcxt);
}
}
}
}
/*
* Are we executing inside a procedure (that is, a nonatomic SPI context)?
*/
bool
SPI_inside_nonatomic_context(void)
{
if (_SPI_current == NULL)
return false; /* not in any SPI context at all */
if (_SPI_current->atomic)
return false; /* it's atomic (ie function not procedure) */
return true;
}
/* Parse, plan, and execute a query string */
int
SPI_execute(const char *src, bool read_only, long tcount)
{
_SPI_plan plan;
SPIExecuteOptions options;
int res;
if (src == NULL || tcount < 0)
return SPI_ERROR_ARGUMENT;
res = _SPI_begin_call(true);
if (res < 0)
return res;
memset(&plan, 0, sizeof(_SPI_plan));
plan.magic = _SPI_PLAN_MAGIC;
plan.parse_mode = RAW_PARSE_DEFAULT;
plan.cursor_options = CURSOR_OPT_PARALLEL_OK;
_SPI_prepare_oneshot_plan(src, &plan);
memset(&options, 0, sizeof(options));
options.read_only = read_only;
options.tcount = tcount;
res = _SPI_execute_plan(&plan, &options,
InvalidSnapshot, InvalidSnapshot,
true);
_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);
}
/* Parse, plan, and execute a query string, with extensible options */
int
SPI_execute_extended(const char *src,
const SPIExecuteOptions *options)
{
int res;
_SPI_plan plan;
if (src == NULL || options == NULL)
return SPI_ERROR_ARGUMENT;
res = _SPI_begin_call(true);
if (res < 0)
return res;
memset(&plan, 0, sizeof(_SPI_plan));
plan.magic = _SPI_PLAN_MAGIC;
plan.parse_mode = RAW_PARSE_DEFAULT;
plan.cursor_options = CURSOR_OPT_PARALLEL_OK;
if (options->params)
{
plan.parserSetup = options->params->parserSetup;
plan.parserSetupArg = options->params->parserSetupArg;
}
_SPI_prepare_oneshot_plan(src, &plan);
res = _SPI_execute_plan(&plan, options,
InvalidSnapshot, InvalidSnapshot,
true);
_SPI_end_call(true);
return res;
}
/* Execute a previously prepared plan */
int
SPI_execute_plan(SPIPlanPtr plan, Datum *Values, const char *Nulls,
bool read_only, long tcount)
{
SPIExecuteOptions options;
int res;
if (plan == NULL || plan->magic != _SPI_PLAN_MAGIC || tcount < 0)
return SPI_ERROR_ARGUMENT;
if (plan->nargs > 0 && Values == NULL)
return SPI_ERROR_PARAM;
res = _SPI_begin_call(true);
if (res < 0)
return res;
memset(&options, 0, sizeof(options));
options.params = _SPI_convert_params(plan->nargs, plan->argtypes,
Values, Nulls);
options.read_only = read_only;
options.tcount = tcount;
res = _SPI_execute_plan(plan, &options,
InvalidSnapshot, InvalidSnapshot,
true);
_SPI_end_call(true);
return res;
}
/* Obsolete version of SPI_execute_plan */
int
SPI_execp(SPIPlanPtr plan, Datum *Values, const char *Nulls, long tcount)
{
return SPI_execute_plan(plan, Values, Nulls, false, tcount);
}
/* Execute a previously prepared plan */
int
SPI_execute_plan_extended(SPIPlanPtr plan,
const SPIExecuteOptions *options)
{
int res;
if (plan == NULL || plan->magic != _SPI_PLAN_MAGIC || options == NULL)
return SPI_ERROR_ARGUMENT;
res = _SPI_begin_call(true);
if (res < 0)
return res;
res = _SPI_execute_plan(plan, options,
InvalidSnapshot, InvalidSnapshot,
true);
_SPI_end_call(true);
return res;
}
/* Execute a previously prepared plan */
int
SPI_execute_plan_with_paramlist(SPIPlanPtr plan, ParamListInfo params,
bool read_only, long tcount)
{
SPIExecuteOptions options;
int res;
if (plan == NULL || plan->magic != _SPI_PLAN_MAGIC || tcount < 0)
return SPI_ERROR_ARGUMENT;
res = _SPI_begin_call(true);
if (res < 0)
return res;
memset(&options, 0, sizeof(options));
options.params = params;
options.read_only = read_only;
options.tcount = tcount;
res = _SPI_execute_plan(plan, &options,
InvalidSnapshot, InvalidSnapshot,
true);
_SPI_end_call(true);
return res;
}
/*
* SPI_execute_snapshot -- identical to SPI_execute_plan, except that we allow
* the caller to specify exactly which snapshots to use, which will be
* registered here. Also, the caller may specify that AFTER triggers should be
* queued as part of the outer query rather than being fired immediately at the
* end of the command.
*
* 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(SPIPlanPtr plan,
Datum *Values, const char *Nulls,
Snapshot snapshot, Snapshot crosscheck_snapshot,
bool read_only, bool fire_triggers, long tcount)
{
SPIExecuteOptions options;
int res;
if (plan == NULL || plan->magic != _SPI_PLAN_MAGIC || tcount < 0)
return SPI_ERROR_ARGUMENT;
if (plan->nargs > 0 && Values == NULL)
return SPI_ERROR_PARAM;
res = _SPI_begin_call(true);
if (res < 0)
return res;
memset(&options, 0, sizeof(options));
options.params = _SPI_convert_params(plan->nargs, plan->argtypes,
Values, Nulls);
options.read_only = read_only;
options.tcount = tcount;
res = _SPI_execute_plan(plan, &options,
snapshot, crosscheck_snapshot,
fire_triggers);
_SPI_end_call(true);
return res;
}
/*
* SPI_execute_with_args -- plan and execute a query with supplied arguments
*
* This is functionally equivalent to SPI_prepare followed by
* SPI_execute_plan.
*/
int
SPI_execute_with_args(const char *src,
int nargs, Oid *argtypes,
Datum *Values, const char *Nulls,
bool read_only, long tcount)
{
int res;
_SPI_plan plan;
ParamListInfo paramLI;
SPIExecuteOptions options;
if (src == NULL || nargs < 0 || tcount < 0)
return SPI_ERROR_ARGUMENT;
if (nargs > 0 && (argtypes == NULL || Values == NULL))
return SPI_ERROR_PARAM;
res = _SPI_begin_call(true);
if (res < 0)
return res;
memset(&plan, 0, sizeof(_SPI_plan));
plan.magic = _SPI_PLAN_MAGIC;
plan.parse_mode = RAW_PARSE_DEFAULT;
plan.cursor_options = CURSOR_OPT_PARALLEL_OK;
plan.nargs = nargs;
plan.argtypes = argtypes;
plan.parserSetup = NULL;
plan.parserSetupArg = NULL;
paramLI = _SPI_convert_params(nargs, argtypes,
Values, Nulls);
_SPI_prepare_oneshot_plan(src, &plan);
memset(&options, 0, sizeof(options));
options.params = paramLI;
options.read_only = read_only;
options.tcount = tcount;
res = _SPI_execute_plan(&plan, &options,
InvalidSnapshot, InvalidSnapshot,
true);
_SPI_end_call(true);
return res;
}
SPIPlanPtr
SPI_prepare(const char *src, int nargs, Oid *argtypes)
{
return SPI_prepare_cursor(src, nargs, argtypes, 0);
}
SPIPlanPtr
SPI_prepare_cursor(const char *src, int nargs, Oid *argtypes,
int cursorOptions)
{
_SPI_plan plan;
SPIPlanPtr 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;
memset(&plan, 0, sizeof(_SPI_plan));
plan.magic = _SPI_PLAN_MAGIC;
plan.parse_mode = RAW_PARSE_DEFAULT;
plan.cursor_options = cursorOptions;
plan.nargs = nargs;
plan.argtypes = argtypes;
plan.parserSetup = NULL;
plan.parserSetupArg = NULL;
_SPI_prepare_plan(src, &plan);
/* copy plan to procedure context */
result = _SPI_make_plan_non_temp(&plan);
_SPI_end_call(true);
return result;
}
SPIPlanPtr
SPI_prepare_extended(const char *src,
const SPIPrepareOptions *options)
{
_SPI_plan plan;
SPIPlanPtr result;
if (src == NULL || options == NULL)
{
SPI_result = SPI_ERROR_ARGUMENT;
return NULL;
}
SPI_result = _SPI_begin_call(true);
if (SPI_result < 0)
return NULL;
memset(&plan, 0, sizeof(_SPI_plan));
plan.magic = _SPI_PLAN_MAGIC;
plan.parse_mode = options->parseMode;
plan.cursor_options = options->cursorOptions;
plan.nargs = 0;
plan.argtypes = NULL;
plan.parserSetup = options->parserSetup;
plan.parserSetupArg = options->parserSetupArg;
_SPI_prepare_plan(src, &plan);
/* copy plan to procedure context */
result = _SPI_make_plan_non_temp(&plan);
_SPI_end_call(true);
return result;
}
SPIPlanPtr
SPI_prepare_params(const char *src,
ParserSetupHook parserSetup,
void *parserSetupArg,
int cursorOptions)
{
_SPI_plan plan;
SPIPlanPtr result;
if (src == NULL)
{
SPI_result = SPI_ERROR_ARGUMENT;
return NULL;
}
SPI_result = _SPI_begin_call(true);
if (SPI_result < 0)
return NULL;
memset(&plan, 0, sizeof(_SPI_plan));
plan.magic = _SPI_PLAN_MAGIC;
plan.parse_mode = RAW_PARSE_DEFAULT;
plan.cursor_options = cursorOptions;
plan.nargs = 0;
plan.argtypes = NULL;
plan.parserSetup = parserSetup;
plan.parserSetupArg = parserSetupArg;
_SPI_prepare_plan(src, &plan);
/* copy plan to procedure context */
result = _SPI_make_plan_non_temp(&plan);
_SPI_end_call(true);
return result;
}
int
SPI_keepplan(SPIPlanPtr plan)
{
ListCell *lc;
if (plan == NULL || plan->magic != _SPI_PLAN_MAGIC ||
plan->saved || plan->oneshot)
return SPI_ERROR_ARGUMENT;
/*
* Mark it saved, reparent it under CacheMemoryContext, and mark all the
* component CachedPlanSources as saved. This sequence cannot fail
* partway through, so there's no risk of long-term memory leakage.
*/
plan->saved = true;
MemoryContextSetParent(plan->plancxt, CacheMemoryContext);
foreach(lc, plan->plancache_list)
{
CachedPlanSource *plansource = (CachedPlanSource *) lfirst(lc);
SaveCachedPlan(plansource);
}
return 0;
}
SPIPlanPtr
SPI_saveplan(SPIPlanPtr plan)
{
SPIPlanPtr newplan;
if (plan == NULL || plan->magic != _SPI_PLAN_MAGIC)
{
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_save_plan(plan);
SPI_result = _SPI_end_call(false);
return newplan;
}
int
SPI_freeplan(SPIPlanPtr plan)
{
ListCell *lc;
if (plan == NULL || plan->magic != _SPI_PLAN_MAGIC)
return SPI_ERROR_ARGUMENT;
/* Release the plancache entries */
foreach(lc, plan->plancache_list)
{
CachedPlanSource *plansource = (CachedPlanSource *) lfirst(lc);
DropCachedPlan(plansource);
}
/* Now get rid of the _SPI_plan and subsidiary data in its plancxt */
MemoryContextDelete(plan->plancxt);
return 0;
}
HeapTuple
SPI_copytuple(HeapTuple tuple)
{
MemoryContext oldcxt;
HeapTuple ctuple;
if (tuple == NULL)
{
SPI_result = SPI_ERROR_ARGUMENT;
return NULL;
}
if (_SPI_current == NULL)
{
SPI_result = SPI_ERROR_UNCONNECTED;
return NULL;
}
oldcxt = MemoryContextSwitchTo(_SPI_current->savedcxt);
ctuple = heap_copytuple(tuple);
MemoryContextSwitchTo(oldcxt);
return ctuple;
}
HeapTupleHeader
SPI_returntuple(HeapTuple tuple, TupleDesc tupdesc)
{
MemoryContext oldcxt;
HeapTupleHeader dtup;
if (tuple == NULL || tupdesc == NULL)
{
SPI_result = SPI_ERROR_ARGUMENT;
return NULL;
}
if (_SPI_current == NULL)
{
SPI_result = SPI_ERROR_UNCONNECTED;
return NULL;
}
/* For RECORD results, make sure a typmod has been assigned */
if (tupdesc->tdtypeid == RECORDOID &&
tupdesc->tdtypmod < 0)
assign_record_type_typmod(tupdesc);
oldcxt = MemoryContextSwitchTo(_SPI_current->savedcxt);
dtup = DatumGetHeapTupleHeader(heap_copy_tuple_as_datum(tuple, tupdesc));
MemoryContextSwitchTo(oldcxt);
return dtup;
}
HeapTuple
SPI_modifytuple(Relation rel, HeapTuple tuple, int natts, int *attnum,
Datum *Values, const char *Nulls)
{
MemoryContext oldcxt;
HeapTuple mtuple;
int numberOfAttributes;
Datum *v;
bool *n;
int i;
if (rel == NULL || tuple == NULL || natts < 0 || attnum == NULL || Values == NULL)
{
SPI_result = SPI_ERROR_ARGUMENT;
return NULL;
}
if (_SPI_current == NULL)
{
SPI_result = SPI_ERROR_UNCONNECTED;
return NULL;
}
oldcxt = MemoryContextSwitchTo(_SPI_current->savedcxt);
SPI_result = 0;
numberOfAttributes = rel->rd_att->natts;
v = (Datum *) palloc(numberOfAttributes * sizeof(Datum));
n = (bool *) palloc(numberOfAttributes * sizeof(bool));
/* fetch old values and nulls */
heap_deform_tuple(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');
}
if (i == natts) /* no errors in *attnum */
{
mtuple = heap_form_tuple(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;
}
else
{
mtuple = NULL;
SPI_result = SPI_ERROR_NOATTRIBUTE;
}
pfree(v);
pfree(n);
MemoryContextSwitchTo(oldcxt);
return mtuple;
}
int
SPI_fnumber(TupleDesc tupdesc, const char *fname)
{
int res;
const FormData_pg_attribute *sysatt;
for (res = 0; res < tupdesc->natts; res++)
{
Form_pg_attribute attr = TupleDescAttr(tupdesc, res);
if (namestrcmp(&attr->attname, fname) == 0 &&
!attr->attisdropped)
return res + 1;
}
sysatt = SystemAttributeByName(fname);
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)
{
const FormData_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 = TupleDescAttr(tupdesc, fnumber - 1);
else
att = SystemAttributeDefinition(fnumber);
return pstrdup(NameStr(att->attname));
}
char *
SPI_getvalue(HeapTuple tuple, TupleDesc tupdesc, int fnumber)
{
Datum val;
bool isnull;
Oid typoid,
foutoid;
bool typisvarlena;
SPI_result = 0;
if (fnumber > tupdesc->natts || fnumber == 0 ||
fnumber <= FirstLowInvalidHeapAttributeNumber)
{
SPI_result = SPI_ERROR_NOATTRIBUTE;
return NULL;
}
val = heap_getattr(tuple, fnumber, tupdesc, &isnull);
if (isnull)
return NULL;
if (fnumber > 0)
typoid = TupleDescAttr(tupdesc, fnumber - 1)->atttypid;
else
typoid = (SystemAttributeDefinition(fnumber))->atttypid;
getTypeOutputInfo(typoid, &foutoid, &typisvarlena);
return OidOutputFunctionCall(foutoid, val);
}
Datum
SPI_getbinval(HeapTuple tuple, TupleDesc tupdesc, int fnumber, bool *isnull)
{
SPI_result = 0;
if (fnumber > tupdesc->natts || 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 = TupleDescAttr(tupdesc, fnumber - 1)->atttypid;
else
typoid = (SystemAttributeDefinition(fnumber))->atttypid;
typeTuple = SearchSysCache1(TYPEOID, ObjectIdGetDatum(typoid));
if (!HeapTupleIsValid(typeTuple))
{
SPI_result = SPI_ERROR_TYPUNKNOWN;
return NULL;
}
result = pstrdup(NameStr(((Form_pg_type) GETSTRUCT(typeTuple))->typname));
ReleaseSysCache(typeTuple);
return result;
}
/*
* Get the data type OID for a column.
*
* There's nothing similar for typmod and typcollation. The rare consumers
* thereof should inspect the TupleDesc directly.
*/
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 TupleDescAttr(tupdesc, fnumber - 1)->atttypid;
else
return (SystemAttributeDefinition(fnumber))->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)
{
if (_SPI_current == NULL)
elog(ERROR, "SPI_palloc called while not connected to SPI");
return MemoryContextAlloc(_SPI_current->savedcxt, size);
}
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);
}
Datum
SPI_datumTransfer(Datum value, bool typByVal, int typLen)
{
MemoryContext oldcxt;
Datum result;
if (_SPI_current == NULL)
elog(ERROR, "SPI_datumTransfer called while not connected to SPI");
oldcxt = MemoryContextSwitchTo(_SPI_current->savedcxt);
result = datumTransfer(value, typByVal, typLen);
MemoryContextSwitchTo(oldcxt);
return result;
}
void
SPI_freetuple(HeapTuple tuple)
{
/* No longer need to worry which context tuple was in... */
heap_freetuple(tuple);
}
void
SPI_freetuptable(SPITupleTable *tuptable)
{
bool found = false;
/* ignore call if NULL pointer */
if (tuptable == NULL)
return;
/*
* Search only the topmost SPI context for a matching tuple table.
*/
if (_SPI_current != NULL)
{
slist_mutable_iter siter;
/* find tuptable in active list, then remove it */
slist_foreach_modify(siter, &_SPI_current->tuptables)
{
SPITupleTable *tt;
tt = slist_container(SPITupleTable, next, siter.cur);
if (tt == tuptable)
{
slist_delete_current(&siter);
found = true;
break;
}
}
}
/*
* Refuse the deletion if we didn't find it in the topmost SPI context.
* This is primarily a guard against double deletion, but might prevent
* other errors as well. Since the worst consequence of not deleting a
* tuptable would be a transient memory leak, this is just a WARNING.
*/
if (!found)
{
elog(WARNING, "attempt to delete invalid SPITupleTable %p", tuptable);
return;
}
/* for safety, reset global variables that might point at tuptable */
if (tuptable == _SPI_current->tuptable)
_SPI_current->tuptable = NULL;
if (tuptable == SPI_tuptable)
SPI_tuptable = NULL;
/* release all memory belonging to tuptable */
MemoryContextDelete(tuptable->tuptabcxt);
}
/*
* SPI_cursor_open()
*
* Open a prepared SPI plan as a portal
*/
Portal
SPI_cursor_open(const char *name, SPIPlanPtr plan,
Datum *Values, const char *Nulls,
bool read_only)
{
Portal portal;
ParamListInfo paramLI;
/* build transient ParamListInfo in caller's context */
paramLI = _SPI_convert_params(plan->nargs, plan->argtypes,
Values, Nulls);
portal = SPI_cursor_open_internal(name, plan, paramLI, read_only);
/* done with the transient ParamListInfo */
if (paramLI)
pfree(paramLI);
return portal;
}
/*
* SPI_cursor_open_with_args()
*
* Parse and plan a query and open it as a portal.
*/
Portal
SPI_cursor_open_with_args(const char *name,
const char *src,
int nargs, Oid *argtypes,
Datum *Values, const char *Nulls,
bool read_only, int cursorOptions)
{
Portal result;
_SPI_plan plan;
ParamListInfo paramLI;
if (src == NULL || nargs < 0)
elog(ERROR, "SPI_cursor_open_with_args called with invalid arguments");
if (nargs > 0 && (argtypes == NULL || Values == NULL))
elog(ERROR, "SPI_cursor_open_with_args called with missing parameters");
SPI_result = _SPI_begin_call(true);
if (SPI_result < 0)
elog(ERROR, "SPI_cursor_open_with_args called while not connected");
memset(&plan, 0, sizeof(_SPI_plan));
plan.magic = _SPI_PLAN_MAGIC;
plan.parse_mode = RAW_PARSE_DEFAULT;
plan.cursor_options = cursorOptions;
plan.nargs = nargs;
plan.argtypes = argtypes;
plan.parserSetup = NULL;
plan.parserSetupArg = NULL;
/* build transient ParamListInfo in executor context */
paramLI = _SPI_convert_params(nargs, argtypes,
Values, Nulls);
_SPI_prepare_plan(src, &plan);
/* We needn't copy the plan; SPI_cursor_open_internal will do so */
result = SPI_cursor_open_internal(name, &plan, paramLI, read_only);
/* And clean up */
_SPI_end_call(true);
return result;
}
/*
* SPI_cursor_open_with_paramlist()
*
* Same as SPI_cursor_open except that parameters (if any) are passed
* as a ParamListInfo, which supports dynamic parameter set determination
*/
Portal
SPI_cursor_open_with_paramlist(const char *name, SPIPlanPtr plan,
ParamListInfo params, bool read_only)
{
return SPI_cursor_open_internal(name, plan, params, read_only);
}
/* Parse a query and open it as a cursor */
Portal
SPI_cursor_parse_open(const char *name,
const char *src,
const SPIParseOpenOptions *options)
{
Portal result;
_SPI_plan plan;
if (src == NULL || options == NULL)
elog(ERROR, "SPI_cursor_parse_open called with invalid arguments");
SPI_result = _SPI_begin_call(true);
if (SPI_result < 0)
elog(ERROR, "SPI_cursor_parse_open called while not connected");
memset(&plan, 0, sizeof(_SPI_plan));
plan.magic = _SPI_PLAN_MAGIC;
plan.parse_mode = RAW_PARSE_DEFAULT;
plan.cursor_options = options->cursorOptions;
if (options->params)
{
plan.parserSetup = options->params->parserSetup;
plan.parserSetupArg = options->params->parserSetupArg;
}
_SPI_prepare_plan(src, &plan);
/* We needn't copy the plan; SPI_cursor_open_internal will do so */
result = SPI_cursor_open_internal(name, &plan,
options->params, options->read_only);
/* And clean up */
_SPI_end_call(true);
return result;
}
/*
* SPI_cursor_open_internal()
*
* Common code for SPI_cursor_open variants
*/
static Portal
SPI_cursor_open_internal(const char *name, SPIPlanPtr plan,
ParamListInfo paramLI, bool read_only)
{
CachedPlanSource *plansource;
CachedPlan *cplan;
List *stmt_list;
char *query_string;
Snapshot snapshot;
MemoryContext oldcontext;
Portal portal;
SPICallbackArg spicallbackarg;
ErrorContextCallback spierrcontext;
/*
* Check that the plan is something the Portal code will special-case as
* returning one tupleset.
*/
if (!SPI_is_cursor_plan(plan))
{
/* try to give a good error message */
const char *cmdtag;
if (list_length(plan->plancache_list) != 1)
ereport(ERROR,
(errcode(ERRCODE_INVALID_CURSOR_DEFINITION),
errmsg("cannot open multi-query plan as cursor")));
plansource = (CachedPlanSource *) linitial(plan->plancache_list);
/* A SELECT that fails SPI_is_cursor_plan() must be SELECT INTO */
if (plansource->commandTag == CMDTAG_SELECT)
cmdtag = "SELECT INTO";
else
cmdtag = GetCommandTagName(plansource->commandTag);
ereport(ERROR,
(errcode(ERRCODE_INVALID_CURSOR_DEFINITION),
/* translator: %s is name of a SQL command, eg INSERT */
errmsg("cannot open %s query as cursor", cmdtag)));
}
Assert(list_length(plan->plancache_list) == 1);
plansource = (CachedPlanSource *) linitial(plan->plancache_list);
/* Push the SPI stack */
if (_SPI_begin_call(true) < 0)
elog(ERROR, "SPI_cursor_open called while not connected");
/* 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);
}
/* Copy the plan's query string into the portal */
query_string = MemoryContextStrdup(portal->portalContext,
plansource->query_string);
/*
* Setup error traceback support for ereport(), in case GetCachedPlan
* throws an error.
*/
spicallbackarg.query = plansource->query_string;
spicallbackarg.mode = plan->parse_mode;
spierrcontext.callback = _SPI_error_callback;
spierrcontext.arg = &spicallbackarg;
spierrcontext.previous = error_context_stack;
error_context_stack = &spierrcontext;
/*
* Note: for a saved plan, we mustn't have any failure occur between
* GetCachedPlan and PortalDefineQuery; that would result in leaking our
* plancache refcount.
*/
/* Replan if needed, and increment plan refcount for portal */
cplan = GetCachedPlan(plansource, paramLI, NULL, _SPI_current->queryEnv);
stmt_list = cplan->stmt_list;
if (!plan->saved)
{
/*
* We don't want the portal to depend on an unsaved CachedPlanSource,
* so must copy the plan into the portal's context. An error here
* will result in leaking our refcount on the plan, but it doesn't
* matter because the plan is unsaved and hence transient anyway.
*/
oldcontext = MemoryContextSwitchTo(portal->portalContext);
stmt_list = copyObject(stmt_list);
MemoryContextSwitchTo(oldcontext);
ReleaseCachedPlan(cplan, NULL);
cplan = NULL; /* portal shouldn't depend on cplan */
}
/*
* Set up the portal.
*/
PortalDefineQuery(portal,
NULL, /* no statement name */
query_string,
plansource->commandTag,
stmt_list,
cplan);
/*
* Set up options for portal. Default SCROLL type is chosen the same way
* as PerformCursorOpen does it.
*/
portal->cursorOptions = plan->cursor_options;
if (!(portal->cursorOptions & (CURSOR_OPT_SCROLL | CURSOR_OPT_NO_SCROLL)))
{
if (list_length(stmt_list) == 1 &&
linitial_node(PlannedStmt, stmt_list)->commandType != CMD_UTILITY &&
linitial_node(PlannedStmt, stmt_list)->rowMarks == NIL &&
ExecSupportsBackwardScan(linitial_node(PlannedStmt, stmt_list)->planTree))
portal->cursorOptions |= CURSOR_OPT_SCROLL;
else
portal->cursorOptions |= CURSOR_OPT_NO_SCROLL;
}
/*
* Disallow SCROLL with SELECT FOR UPDATE. This is not redundant with the
* check in transformDeclareCursorStmt because the cursor options might
* not have come through there.
*/
if (portal->cursorOptions & CURSOR_OPT_SCROLL)
{
if (list_length(stmt_list) == 1 &&
linitial_node(PlannedStmt, stmt_list)->commandType != CMD_UTILITY &&
linitial_node(PlannedStmt, stmt_list)->rowMarks != NIL)
ereport(ERROR,
(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
errmsg("DECLARE SCROLL CURSOR ... FOR UPDATE/SHARE is not supported"),
errdetail("Scrollable cursors must be READ ONLY.")));
}
/* Make current query environment available to portal at execution time. */
portal->queryEnv = _SPI_current->queryEnv;
/*
* If told to be read-only, we'd better check for read-only queries. This
* can't be done earlier because we need to look at the finished, planned
* queries. (In particular, we don't want to do it between GetCachedPlan
* and PortalDefineQuery, because throwing an error between those steps
* would result in leaking our plancache refcount.)
*/
if (read_only)
{
ListCell *lc;
foreach(lc, stmt_list)
{
PlannedStmt *pstmt = lfirst_node(PlannedStmt, lc);
if (!CommandIsReadOnly(pstmt))
ereport(ERROR,
(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
/* translator: %s is a SQL statement name */
errmsg("%s is not allowed in a non-volatile function",
CreateCommandName((Node *) pstmt))));
}
}
/* Set up the snapshot to use. */
if (read_only)
snapshot = GetActiveSnapshot();
else
{
CommandCounterIncrement();
snapshot = GetTransactionSnapshot();
}
/*
* If the plan has parameters, copy them into the portal. Note that this
* must be done after revalidating the plan, because in dynamic parameter
* cases the set of parameters could have changed during re-parsing.
*/
if (paramLI)
{
oldcontext = MemoryContextSwitchTo(portal->portalContext);
paramLI = copyParamList(paramLI);
MemoryContextSwitchTo(oldcontext);
}
/*
* Start portal execution.
*/
PortalStart(portal, paramLI, 0, snapshot);
Assert(portal->strategy != PORTAL_MULTI_QUERY);
/* Pop the error context stack */
error_context_stack = spierrcontext.previous;
/* Pop the SPI stack */
_SPI_end_call(true);
/* 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 ? FETCH_FORWARD : FETCH_BACKWARD, count,
CreateDestReceiver(DestSPI));
/* 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 ? FETCH_FORWARD : FETCH_BACKWARD, count,
None_Receiver);
}
/*
* SPI_scroll_cursor_fetch()
*
* Fetch rows in a scrollable cursor
*/
void
SPI_scroll_cursor_fetch(Portal portal, FetchDirection direction, long count)
{
_SPI_cursor_operation(portal,
direction, count,
CreateDestReceiver(DestSPI));
/* we know that the DestSPI receiver doesn't need a destroy call */
}
/*
* SPI_scroll_cursor_move()
*
* Move in a scrollable cursor
*/
void
SPI_scroll_cursor_move(Portal portal, FetchDirection direction, long count)
{
_SPI_cursor_operation(portal, direction, 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(SPIPlanPtr plan, int argIndex)
{
if (plan == NULL || plan->magic != _SPI_PLAN_MAGIC ||
argIndex < 0 || argIndex >= plan->nargs)
{
SPI_result = SPI_ERROR_ARGUMENT;
return InvalidOid;
}
return plan->argtypes[argIndex];
}
/*
* Returns the number of arguments for the prepared plan.
*/
int
SPI_getargcount(SPIPlanPtr plan)
{
if (plan == NULL || plan->magic != _SPI_PLAN_MAGIC)
{
SPI_result = SPI_ERROR_ARGUMENT;
return -1;
}
return 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(SPIPlanPtr plan)
{
CachedPlanSource *plansource;
if (plan == NULL || plan->magic != _SPI_PLAN_MAGIC)
{
SPI_result = SPI_ERROR_ARGUMENT;
return false;
}
if (list_length(plan->plancache_list) != 1)
{
SPI_result = 0;
return false; /* not exactly 1 pre-rewrite command */
}
plansource = (CachedPlanSource *) linitial(plan->plancache_list);
/*
* We used to force revalidation of the cached plan here, but that seems
* unnecessary: invalidation could mean a change in the rowtype of the
* tuples returned by a plan, but not whether it returns tuples at all.
*/
SPI_result = 0;
/* Does it return tuples? */
if (plansource->resultDesc)
return true;
return false;
}
/*
* SPI_plan_is_valid --- test whether a SPI plan is currently valid
* (that is, not marked as being in need of revalidation).
*
* See notes for CachedPlanIsValid before using this.
*/
bool
SPI_plan_is_valid(SPIPlanPtr plan)
{
ListCell *lc;
Assert(plan->magic == _SPI_PLAN_MAGIC);
foreach(lc, plan->plancache_list)
{
CachedPlanSource *plansource = (CachedPlanSource *) lfirst(lc);
if (!CachedPlanIsValid(plansource))
return false;
}
return true;
}
/*
* 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_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_ERROR_REL_DUPLICATE:
return "SPI_ERROR_REL_DUPLICATE";
case SPI_ERROR_REL_NOT_FOUND:
return "SPI_ERROR_REL_NOT_FOUND";
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";
case SPI_OK_REWRITTEN:
return "SPI_OK_REWRITTEN";
case SPI_OK_REL_REGISTER:
return "SPI_OK_REL_REGISTER";
case SPI_OK_REL_UNREGISTER:
return "SPI_OK_REL_UNREGISTER";
case SPI_OK_TD_REGISTER:
return "SPI_OK_TD_REGISTER";
case SPI_OK_MERGE:
return "SPI_OK_MERGE";
case SPI_OK_MERGE_RETURNING:
return "SPI_OK_MERGE_RETURNING";
}
/* Unrecognized code ... return something useful ... */
sprintf(buf, "Unrecognized SPI code %d", code);
return buf;
}
/*
* SPI_plan_get_plan_sources --- get a SPI plan's underlying list of
* CachedPlanSources.
*
* This is exported so that PL/pgSQL can use it (this beats letting PL/pgSQL
* look directly into the SPIPlan for itself). It's not documented in
* spi.sgml because we'd just as soon not have too many places using this.
*/
List *
SPI_plan_get_plan_sources(SPIPlanPtr plan)
{
Assert(plan->magic == _SPI_PLAN_MAGIC);
return plan->plancache_list;
}
/*
* SPI_plan_get_cached_plan --- get a SPI plan's generic CachedPlan,
* if the SPI plan contains exactly one CachedPlanSource. If not,
* return NULL.
*
* The plan's refcount is incremented (and logged in CurrentResourceOwner,
* if it's a saved plan). Caller is responsible for doing ReleaseCachedPlan.
*
* This is exported so that PL/pgSQL can use it (this beats letting PL/pgSQL
* look directly into the SPIPlan for itself). It's not documented in
* spi.sgml because we'd just as soon not have too many places using this.
*/
CachedPlan *
SPI_plan_get_cached_plan(SPIPlanPtr plan)
{
CachedPlanSource *plansource;
CachedPlan *cplan;
SPICallbackArg spicallbackarg;
ErrorContextCallback spierrcontext;
Assert(plan->magic == _SPI_PLAN_MAGIC);
/* Can't support one-shot plans here */
if (plan->oneshot)
return NULL;
/* Must have exactly one CachedPlanSource */
if (list_length(plan->plancache_list) != 1)
return NULL;
plansource = (CachedPlanSource *) linitial(plan->plancache_list);
/* Setup error traceback support for ereport() */
spicallbackarg.query = plansource->query_string;
spicallbackarg.mode = plan->parse_mode;
spierrcontext.callback = _SPI_error_callback;
spierrcontext.arg = &spicallbackarg;
spierrcontext.previous = error_context_stack;
error_context_stack = &spierrcontext;
/* Get the generic plan for the query */
cplan = GetCachedPlan(plansource, NULL,
plan->saved ? CurrentResourceOwner : NULL,
_SPI_current->queryEnv);
Assert(cplan == plansource->gplan);
/* Pop the error context stack */
error_context_stack = spierrcontext.previous;
return cplan;
}
/* =================== 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;
if (_SPI_current == NULL)
elog(ERROR, "spi_dest_startup called while not connected to SPI");
if (_SPI_current->tuptable != NULL)
elog(ERROR, "improper call to spi_dest_startup");
/* We create the tuple table context as a child of procCxt */
oldcxt = _SPI_procmem(); /* switch to procedure memory context */
tuptabcxt = AllocSetContextCreate(CurrentMemoryContext,
"SPI TupTable",
ALLOCSET_DEFAULT_SIZES);
MemoryContextSwitchTo(tuptabcxt);
_SPI_current->tuptable = tuptable = (SPITupleTable *)
palloc0(sizeof(SPITupleTable));
tuptable->tuptabcxt = tuptabcxt;
tuptable->subid = GetCurrentSubTransactionId();
/*
* The tuptable is now valid enough to be freed by AtEOSubXact_SPI, so put
* it onto the SPI context's tuptables list. This will ensure it's not
* leaked even in the unlikely event the following few lines fail.
*/
slist_push_head(&_SPI_current->tuptables, &tuptable->next);
/* set up initial allocations */
tuptable->alloced = 128;
tuptable->vals = (HeapTuple *) palloc(tuptable->alloced * sizeof(HeapTuple));
tuptable->numvals = 0;
tuptable->tupdesc = CreateTupleDescCopy(typeinfo);
MemoryContextSwitchTo(oldcxt);
}
/*
* spi_printtup
* store tuple retrieved by Executor into SPITupleTable
* of current SPI procedure
*/
bool
spi_printtup(TupleTableSlot *slot, DestReceiver *self)
{
SPITupleTable *tuptable;
MemoryContext oldcxt;
if (_SPI_current == NULL)
elog(ERROR, "spi_printtup called while not connected to SPI");
tuptable = _SPI_current->tuptable;
if (tuptable == NULL)
elog(ERROR, "improper call to spi_printtup");
oldcxt = MemoryContextSwitchTo(tuptable->tuptabcxt);
if (tuptable->numvals >= tuptable->alloced)
{
/* Double the size of the pointer array */
uint64 newalloced = tuptable->alloced * 2;
tuptable->vals = (HeapTuple *) repalloc_huge(tuptable->vals,
newalloced * sizeof(HeapTuple));
tuptable->alloced = newalloced;
}
tuptable->vals[tuptable->numvals] = ExecCopySlotHeapTuple(slot);
(tuptable->numvals)++;
MemoryContextSwitchTo(oldcxt);
return true;
}
/*
* Static functions
*/
/*
* Parse and analyze a querystring.
*
* At entry, plan->argtypes and plan->nargs (or alternatively plan->parserSetup
* and plan->parserSetupArg) must be valid, as must plan->parse_mode and
* plan->cursor_options.
*
* Results are stored into *plan (specifically, plan->plancache_list).
* Note that the result data is all in CurrentMemoryContext or child contexts
* thereof; in practice this means it is in the SPI executor context, and
* what we are creating is a "temporary" SPIPlan. Cruft generated during
* parsing is also left in CurrentMemoryContext.
*/
static void
_SPI_prepare_plan(const char *src, SPIPlanPtr plan)
{
List *raw_parsetree_list;
List *plancache_list;
ListCell *list_item;
SPICallbackArg spicallbackarg;
ErrorContextCallback spierrcontext;
/*
* Setup error traceback support for ereport()
*/
spicallbackarg.query = src;
spicallbackarg.mode = plan->parse_mode;
spierrcontext.callback = _SPI_error_callback;
spierrcontext.arg = &spicallbackarg;
spierrcontext.previous = error_context_stack;
error_context_stack = &spierrcontext;
/*
* Parse the request string into a list of raw parse trees.
*/
raw_parsetree_list = raw_parser(src, plan->parse_mode);
/*
* Do parse analysis and rule rewrite for each raw parsetree, storing the
* results into unsaved plancache entries.
*/
plancache_list = NIL;
foreach(list_item, raw_parsetree_list)
{
RawStmt *parsetree = lfirst_node(RawStmt, list_item);
List *stmt_list;
CachedPlanSource *plansource;
/*
* Create the CachedPlanSource before we do parse analysis, since it
* needs to see the unmodified raw parse tree.
*/
plansource = CreateCachedPlan(parsetree,
src,
CreateCommandTag(parsetree->stmt));
/*
* Parameter datatypes are driven by parserSetup hook if provided,
* otherwise we use the fixed parameter list.
*/
if (plan->parserSetup != NULL)
{
Assert(plan->nargs == 0);
stmt_list = pg_analyze_and_rewrite_withcb(parsetree,
src,
plan->parserSetup,
plan->parserSetupArg,
_SPI_current->queryEnv);
}
else
{
stmt_list = pg_analyze_and_rewrite_fixedparams(parsetree,
src,
plan->argtypes,
plan->nargs,
_SPI_current->queryEnv);
}
/* Finish filling in the CachedPlanSource */
CompleteCachedPlan(plansource,
stmt_list,
NULL,
plan->argtypes,
plan->nargs,
plan->parserSetup,
plan->parserSetupArg,
plan->cursor_options,
false); /* not fixed result */
plancache_list = lappend(plancache_list, plansource);
}
plan->plancache_list = plancache_list;
plan->oneshot = false;
/*
* Pop the error context stack
*/
error_context_stack = spierrcontext.previous;
}
/*
* Parse, but don't analyze, a querystring.
*
* This is a stripped-down version of _SPI_prepare_plan that only does the
* initial raw parsing. It creates "one shot" CachedPlanSources
* that still require parse analysis before execution is possible.
*
* The advantage of using the "one shot" form of CachedPlanSource is that
* we eliminate data copying and invalidation overhead. Postponing parse
* analysis also prevents issues if some of the raw parsetrees are DDL
* commands that affect validity of later parsetrees. Both of these
* attributes are good things for SPI_execute() and similar cases.
*
* Results are stored into *plan (specifically, plan->plancache_list).
* Note that the result data is all in CurrentMemoryContext or child contexts
* thereof; in practice this means it is in the SPI executor context, and
* what we are creating is a "temporary" SPIPlan. Cruft generated during
* parsing is also left in CurrentMemoryContext.
*/
static void
_SPI_prepare_oneshot_plan(const char *src, SPIPlanPtr plan)
{
List *raw_parsetree_list;
List *plancache_list;
ListCell *list_item;
SPICallbackArg spicallbackarg;
ErrorContextCallback spierrcontext;
/*
* Setup error traceback support for ereport()
*/
spicallbackarg.query = src;
spicallbackarg.mode = plan->parse_mode;
spierrcontext.callback = _SPI_error_callback;
spierrcontext.arg = &spicallbackarg;
spierrcontext.previous = error_context_stack;
error_context_stack = &spierrcontext;
/*
* Parse the request string into a list of raw parse trees.
*/
raw_parsetree_list = raw_parser(src, plan->parse_mode);
/*
* Construct plancache entries, but don't do parse analysis yet.
*/
plancache_list = NIL;
foreach(list_item, raw_parsetree_list)
{
RawStmt *parsetree = lfirst_node(RawStmt, list_item);
CachedPlanSource *plansource;
plansource = CreateOneShotCachedPlan(parsetree,
src,
CreateCommandTag(parsetree->stmt));
plancache_list = lappend(plancache_list, plansource);
}
plan->plancache_list = plancache_list;
plan->oneshot = true;
/*
* Pop the error context stack
*/
error_context_stack = spierrcontext.previous;
}
/*
* _SPI_execute_plan: execute the given plan with the given options
*
* options contains options accessible from outside SPI:
* params: parameter values to pass to query
* read_only: true for read-only execution (no CommandCounterIncrement)
* allow_nonatomic: true to allow nonatomic CALL/DO execution
* must_return_tuples: throw error if query doesn't return tuples
* tcount: execution tuple-count limit, or 0 for none
* dest: DestReceiver to receive output, or NULL for normal SPI output
* owner: ResourceOwner that will be used to hold refcount on plan;
* if NULL, CurrentResourceOwner is used (ignored for non-saved plan)
*
* Additional, only-internally-accessible options:
* 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
* fire_triggers: true to fire AFTER triggers at end of query (normal case);
* false means any AFTER triggers are postponed to end of outer query
*/
static int
_SPI_execute_plan(SPIPlanPtr plan, const SPIExecuteOptions *options,
Snapshot snapshot, Snapshot crosscheck_snapshot,
bool fire_triggers)
{
int my_res = 0;
uint64 my_processed = 0;
SPITupleTable *my_tuptable = NULL;
int res = 0;
bool pushed_active_snap = false;
ResourceOwner plan_owner = options->owner;
SPICallbackArg spicallbackarg;
ErrorContextCallback spierrcontext;
CachedPlan *cplan = NULL;
ListCell *lc1;
/*
* Setup error traceback support for ereport()
*/
spicallbackarg.query = NULL; /* we'll fill this below */
spicallbackarg.mode = plan->parse_mode;
spierrcontext.callback = _SPI_error_callback;
spierrcontext.arg = &spicallbackarg;
spierrcontext.previous = error_context_stack;
error_context_stack = &spierrcontext;
/*
* We support four distinct snapshot management behaviors:
*
* snapshot != InvalidSnapshot, read_only = true: use exactly the given
* snapshot.
*
* snapshot != InvalidSnapshot, read_only = false: use the given snapshot,
* modified by advancing its command ID before each querytree.
*
* snapshot == InvalidSnapshot, read_only = true: use the entry-time
* ActiveSnapshot, if any (if there isn't one, we run with no snapshot).
*
* snapshot == InvalidSnapshot, read_only = false: take a full new
* snapshot for each user command, and advance its command ID before each
* querytree within the command.
*
* In the first two cases, we can just push the snap onto the stack once
* for the whole plan list.
*
* Note that snapshot != InvalidSnapshot implies an atomic execution
* context.
*/
if (snapshot != InvalidSnapshot)
{
Assert(!options->allow_nonatomic);
if (options->read_only)
{
PushActiveSnapshot(snapshot);
pushed_active_snap = true;
}
else
{
/* Make sure we have a private copy of the snapshot to modify */
PushCopiedSnapshot(snapshot);
pushed_active_snap = true;
}
}
/*
* Ensure that we have a resource owner if plan is saved, and not if it
* isn't.
*/
if (!plan->saved)
plan_owner = NULL;
else if (plan_owner == NULL)
plan_owner = CurrentResourceOwner;
/*
* We interpret must_return_tuples as "there must be at least one query,
* and all of them must return tuples". This is a bit laxer than
* SPI_is_cursor_plan's check, but there seems no reason to enforce that
* there be only one query.
*/
if (options->must_return_tuples && plan->plancache_list == NIL)
ereport(ERROR,
(errcode(ERRCODE_SYNTAX_ERROR),
errmsg("empty query does not return tuples")));
foreach(lc1, plan->plancache_list)
{
CachedPlanSource *plansource = (CachedPlanSource *) lfirst(lc1);
List *stmt_list;
ListCell *lc2;
spicallbackarg.query = plansource->query_string;
/*
* If this is a one-shot plan, we still need to do parse analysis.
*/
if (plan->oneshot)
{
RawStmt *parsetree = plansource->raw_parse_tree;
const char *src = plansource->query_string;
List *querytree_list;
/*
* Parameter datatypes are driven by parserSetup hook if provided,
* otherwise we use the fixed parameter list.
*/
if (parsetree == NULL)
querytree_list = NIL;
else if (plan->parserSetup != NULL)
{
Assert(plan->nargs == 0);
querytree_list = pg_analyze_and_rewrite_withcb(parsetree,
src,
plan->parserSetup,
plan->parserSetupArg,
_SPI_current->queryEnv);
}
else
{
querytree_list = pg_analyze_and_rewrite_fixedparams(parsetree,
src,
plan->argtypes,
plan->nargs,
_SPI_current->queryEnv);
}
/* Finish filling in the CachedPlanSource */
CompleteCachedPlan(plansource,
querytree_list,
NULL,
plan->argtypes,
plan->nargs,
plan->parserSetup,
plan->parserSetupArg,
plan->cursor_options,
false); /* not fixed result */
}
/*
* If asked to, complain when query does not return tuples.
* (Replanning can't change this, so we can check it before that.
* However, we can't check it till after parse analysis, so in the
* case of a one-shot plan this is the earliest we could check.)
*/
if (options->must_return_tuples && !plansource->resultDesc)
{
/* try to give a good error message */
const char *cmdtag;
/* A SELECT without resultDesc must be SELECT INTO */
if (plansource->commandTag == CMDTAG_SELECT)
cmdtag = "SELECT INTO";
else
cmdtag = GetCommandTagName(plansource->commandTag);
ereport(ERROR,
(errcode(ERRCODE_SYNTAX_ERROR),
/* translator: %s is name of a SQL command, eg INSERT */
errmsg("%s query does not return tuples", cmdtag)));
}
/*
* Replan if needed, and increment plan refcount. If it's a saved
* plan, the refcount must be backed by the plan_owner.
*/
cplan = GetCachedPlan(plansource, options->params,
plan_owner, _SPI_current->queryEnv);
stmt_list = cplan->stmt_list;
/*
* If we weren't given a specific snapshot to use, and the statement
* list requires a snapshot, set that up.
*/
if (snapshot == InvalidSnapshot &&
(list_length(stmt_list) > 1 ||
(list_length(stmt_list) == 1 &&
PlannedStmtRequiresSnapshot(linitial_node(PlannedStmt,
stmt_list)))))
{
/*
* First, ensure there's a Portal-level snapshot. This back-fills
* the snapshot stack in case the previous operation was a COMMIT
* or ROLLBACK inside a procedure or DO block. (We can't put back
* the Portal snapshot any sooner, or we'd break cases like doing
* SET or LOCK just after COMMIT.) It's enough to check once per
* statement list, since COMMIT/ROLLBACK/CALL/DO can't appear
* within a multi-statement list.
*/
EnsurePortalSnapshotExists();
/*
* In the default non-read-only case, get a new per-statement-list
* snapshot, replacing any that we pushed in a previous cycle.
* Skip it when doing non-atomic execution, though (we rely
* entirely on the Portal snapshot in that case).
*/
if (!options->read_only && !options->allow_nonatomic)
{
if (pushed_active_snap)
PopActiveSnapshot();
PushActiveSnapshot(GetTransactionSnapshot());
pushed_active_snap = true;
}
}
foreach(lc2, stmt_list)
{
PlannedStmt *stmt = lfirst_node(PlannedStmt, lc2);
bool canSetTag = stmt->canSetTag;
DestReceiver *dest;
/*
* Reset output state. (Note that if a non-SPI receiver is used,
* _SPI_current->processed will stay zero, and that's what we'll
* report to the caller. It's the receiver's job to count tuples
* in that case.)
*/
_SPI_current->processed = 0;
_SPI_current->tuptable = NULL;
/* Check for unsupported cases. */
if (stmt->utilityStmt)
{
if (IsA(stmt->utilityStmt, CopyStmt))
{
CopyStmt *cstmt = (CopyStmt *) stmt->utilityStmt;
if (cstmt->filename == NULL)
{
my_res = SPI_ERROR_COPY;
goto fail;
}
}
else if (IsA(stmt->utilityStmt, TransactionStmt))
{
my_res = SPI_ERROR_TRANSACTION;
goto fail;
}
}
if (options->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",
CreateCommandName((Node *) stmt))));
/*
* If not read-only mode, advance the command counter before each
* command and update the snapshot. (But skip it if the snapshot
* isn't under our control.)
*/
if (!options->read_only && pushed_active_snap)
{
CommandCounterIncrement();
UpdateActiveSnapshotCommandId();
}
/*
* Select appropriate tuple receiver. Output from non-canSetTag
* subqueries always goes to the bit bucket.
*/
if (!canSetTag)
dest = CreateDestReceiver(DestNone);
else if (options->dest)
dest = options->dest;
else
dest = CreateDestReceiver(DestSPI);
if (stmt->utilityStmt == NULL)
{
QueryDesc *qdesc;
Snapshot snap;
if (ActiveSnapshotSet())
snap = GetActiveSnapshot();
else
snap = InvalidSnapshot;
qdesc = CreateQueryDesc(stmt,
plansource->query_string,
snap, crosscheck_snapshot,
dest,
options->params,
_SPI_current->queryEnv,
0);
res = _SPI_pquery(qdesc, fire_triggers,
canSetTag ? options->tcount : 0);
FreeQueryDesc(qdesc);
}
else
{
ProcessUtilityContext context;
QueryCompletion qc;
/*
* If the SPI context is atomic, or we were not told to allow
* nonatomic operations, tell ProcessUtility this is an atomic
* execution context.
*/
if (_SPI_current->atomic || !options->allow_nonatomic)
context = PROCESS_UTILITY_QUERY;
else
context = PROCESS_UTILITY_QUERY_NONATOMIC;
InitializeQueryCompletion(&qc);
ProcessUtility(stmt,
plansource->query_string,
true, /* protect plancache's node tree */
context,
options->params,
_SPI_current->queryEnv,
dest,
&qc);
/* Update "processed" if stmt returned tuples */
if (_SPI_current->tuptable)
_SPI_current->processed = _SPI_current->tuptable->numvals;
res = SPI_OK_UTILITY;
/*
* Some utility statements return a row count, even though the
* tuples are not returned to the caller.
*/
if (IsA(stmt->utilityStmt, CreateTableAsStmt))
{
CreateTableAsStmt *ctastmt = (CreateTableAsStmt *) stmt->utilityStmt;
if (qc.commandTag == CMDTAG_SELECT)
_SPI_current->processed = qc.nprocessed;
else
{
/*
* Must be an IF NOT EXISTS that did nothing, or a
* CREATE ... WITH NO DATA.
*/
Assert(ctastmt->if_not_exists ||
ctastmt->into->skipData);
_SPI_current->processed = 0;
}
/*
* For historical reasons, if CREATE TABLE AS was spelled
* as SELECT INTO, return a special return code.
*/
if (ctastmt->is_select_into)
res = SPI_OK_SELINTO;
}
else if (IsA(stmt->utilityStmt, CopyStmt))
{
Assert(qc.commandTag == CMDTAG_COPY);
_SPI_current->processed = qc.nprocessed;
}
}
/*
* The last canSetTag query sets the status values returned to the
* caller. Be careful to free any tuptables not returned, to
* avoid intra-transaction memory leak.
*/
if (canSetTag)
{
my_processed = _SPI_current->processed;
SPI_freetuptable(my_tuptable);
my_tuptable = _SPI_current->tuptable;
my_res = res;
}
else
{
SPI_freetuptable(_SPI_current->tuptable);
_SPI_current->tuptable = NULL;
}
/*
* We don't issue a destroy call to the receiver. The SPI and
* None receivers would ignore it anyway, while if the caller
* supplied a receiver, it's not our job to destroy it.
*/
if (res < 0)
{
my_res = res;
goto fail;
}
}
/* Done with this plan, so release refcount */
ReleaseCachedPlan(cplan, plan_owner);
cplan = NULL;
/*
* If not read-only mode, advance the command counter after the last
* command. This ensures that its effects are visible, in case it was
* DDL that would affect the next CachedPlanSource.
*/
if (!options->read_only)
CommandCounterIncrement();
}
fail:
/* Pop the snapshot off the stack if we pushed one */
if (pushed_active_snap)
PopActiveSnapshot();
/* We no longer need the cached plan refcount, if any */
if (cplan)
ReleaseCachedPlan(cplan, plan_owner);
/*
* Pop the error context stack
*/
error_context_stack = spierrcontext.previous;
/* Save results for caller */
SPI_processed = my_processed;
SPI_tuptable = my_tuptable;
/* tuptable now is caller's responsibility, not SPI's */
_SPI_current->tuptable = NULL;
/*
* If none of the queries had canSetTag, return SPI_OK_REWRITTEN. Prior to
* 8.4, we used return the last query's result code, but not its auxiliary
* results, but that's confusing.
*/
if (my_res == 0)
my_res = SPI_OK_REWRITTEN;
return my_res;
}
/*
* Convert arrays of query parameters to form wanted by planner and executor
*/
static ParamListInfo
_SPI_convert_params(int nargs, Oid *argtypes,
Datum *Values, const char *Nulls)
{
ParamListInfo paramLI;
if (nargs > 0)
{
paramLI = makeParamList(nargs);
for (int i = 0; i < nargs; i++)
{
ParamExternData *prm = &paramLI->params[i];
prm->value = Values[i];
prm->isnull = (Nulls && Nulls[i] == 'n');
prm->pflags = PARAM_FLAG_CONST;
prm->ptype = argtypes[i];
}
}
else
paramLI = NULL;
return paramLI;
}
static int
_SPI_pquery(QueryDesc *queryDesc, bool fire_triggers, uint64 tcount)
{
int operation = queryDesc->operation;
int eflags;
int res;
switch (operation)
{
case CMD_SELECT:
if (queryDesc->dest->mydest == DestNone)
{
/* 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->hasReturning)
res = SPI_OK_INSERT_RETURNING;
else
res = SPI_OK_INSERT;
break;
case CMD_DELETE:
if (queryDesc->plannedstmt->hasReturning)
res = SPI_OK_DELETE_RETURNING;
else
res = SPI_OK_DELETE;
break;
case CMD_UPDATE:
if (queryDesc->plannedstmt->hasReturning)
res = SPI_OK_UPDATE_RETURNING;
else
res = SPI_OK_UPDATE;
break;
case CMD_MERGE:
if (queryDesc->plannedstmt->hasReturning)
res = SPI_OK_MERGE_RETURNING;
else
res = SPI_OK_MERGE;
break;
default:
return SPI_ERROR_OPUNKNOWN;
}
#ifdef SPI_EXECUTOR_STATS
if (ShowExecutorStats)
ResetUsage();
#endif
/* Select execution options */
if (fire_triggers)
eflags = 0; /* default run-to-completion flags */
else
eflags = EXEC_FLAG_SKIP_TRIGGERS;
ExecutorStart(queryDesc, eflags);
ExecutorRun(queryDesc, ForwardScanDirection, tcount, true);
_SPI_current->processed = queryDesc->estate->es_processed;
if ((res == SPI_OK_SELECT || queryDesc->plannedstmt->hasReturning) &&
queryDesc->dest->mydest == DestSPI)
{
if (_SPI_checktuples())
elog(ERROR, "consistency check on SPI tuple count failed");
}
ExecutorFinish(queryDesc);
ExecutorEnd(queryDesc);
/* FreeQueryDesc is done by the caller */
#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)
{
SPICallbackArg *carg = (SPICallbackArg *) arg;
const char *query = carg->query;
int syntaxerrposition;
if (query == NULL) /* in case arg wasn't set yet */
return;
/*
* 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
{
/* Use the parse mode to decide how to describe the query */
switch (carg->mode)
{
case RAW_PARSE_PLPGSQL_EXPR:
errcontext("SQL expression \"%s\"", query);
break;
case RAW_PARSE_PLPGSQL_ASSIGN1:
case RAW_PARSE_PLPGSQL_ASSIGN2:
case RAW_PARSE_PLPGSQL_ASSIGN3:
errcontext("PL/pgSQL assignment \"%s\"", query);
break;
default:
errcontext("SQL statement \"%s\"", query);
break;
}
}
}
/*
* _SPI_cursor_operation()
*
* Do a FETCH or MOVE in a cursor
*/
static void
_SPI_cursor_operation(Portal portal, FetchDirection direction, long count,
DestReceiver *dest)
{
uint64 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,
direction,
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
*
* use_exec is true if we intend to make use of the procedure's execCxt
* during this SPI operation. We'll switch into that context, and arrange
* for it to be cleaned up at _SPI_end_call or if an error occurs.
*/
static int
_SPI_begin_call(bool use_exec)
{
if (_SPI_current == NULL)
return SPI_ERROR_UNCONNECTED;
if (use_exec)
{
/* remember when the Executor operation started */
_SPI_current->execSubid = GetCurrentSubTransactionId();
/* switch to the Executor memory context */
_SPI_execmem();
}
return 0;
}
/*
* _SPI_end_call: end a SPI operation within a connected procedure
*
* use_exec must be the same as in the previous _SPI_begin_call
*
* Note: this currently has no failure return cases, so callers don't check
*/
static int
_SPI_end_call(bool use_exec)
{
if (use_exec)
{
/* switch to the procedure memory context */
_SPI_procmem();
/* mark Executor context no longer in use */
_SPI_current->execSubid = InvalidSubTransactionId;
/* and free Executor memory */
MemoryContextReset(_SPI_current->execCxt);
}
return 0;
}
static bool
_SPI_checktuples(void)
{
uint64 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->numvals)
failed = true;
return failed;
}
/*
* Convert a "temporary" SPIPlan into an "unsaved" plan.
*
* The passed _SPI_plan struct is on the stack, and all its subsidiary data
* is in or under the current SPI executor context. Copy the plan into the
* SPI procedure context so it will survive _SPI_end_call(). To minimize
* data copying, this destructively modifies the input plan, by taking the
* plancache entries away from it and reparenting them to the new SPIPlan.
*/
static SPIPlanPtr
_SPI_make_plan_non_temp(SPIPlanPtr plan)
{
SPIPlanPtr newplan;
MemoryContext parentcxt = _SPI_current->procCxt;
MemoryContext plancxt;
MemoryContext oldcxt;
ListCell *lc;
/* Assert the input is a temporary SPIPlan */
Assert(plan->magic == _SPI_PLAN_MAGIC);
Assert(plan->plancxt == NULL);
/* One-shot plans can't be saved */
Assert(!plan->oneshot);
/*
* Create a memory context for the plan, underneath the procedure context.
* We don't expect the plan to be very large.
*/
plancxt = AllocSetContextCreate(parentcxt,
"SPI Plan",
ALLOCSET_SMALL_SIZES);
oldcxt = MemoryContextSwitchTo(plancxt);
/* Copy the _SPI_plan struct and subsidiary data into the new context */
newplan = (SPIPlanPtr) palloc0(sizeof(_SPI_plan));
newplan->magic = _SPI_PLAN_MAGIC;
newplan->plancxt = plancxt;
newplan->parse_mode = plan->parse_mode;
newplan->cursor_options = plan->cursor_options;
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;
newplan->parserSetup = plan->parserSetup;
newplan->parserSetupArg = plan->parserSetupArg;
/*
* Reparent all the CachedPlanSources into the procedure context. In
* theory this could fail partway through due to the pallocs, but we don't
* care too much since both the procedure context and the executor context
* would go away on error.
*/
foreach(lc, plan->plancache_list)
{
CachedPlanSource *plansource = (CachedPlanSource *) lfirst(lc);
CachedPlanSetParentContext(plansource, parentcxt);
/* Build new list, with list cells in plancxt */
newplan->plancache_list = lappend(newplan->plancache_list, plansource);
}
MemoryContextSwitchTo(oldcxt);
/* For safety, unlink the CachedPlanSources from the temporary plan */
plan->plancache_list = NIL;
return newplan;
}
/*
* Make a "saved" copy of the given plan.
*/
static SPIPlanPtr
_SPI_save_plan(SPIPlanPtr plan)
{
SPIPlanPtr newplan;
MemoryContext plancxt;
MemoryContext oldcxt;
ListCell *lc;
/* One-shot plans can't be saved */
Assert(!plan->oneshot);
/*
* Create a memory context for the plan. We don't expect the plan to be
* very large, so use smaller-than-default alloc parameters. It's a
* transient context until we finish copying everything.
*/
plancxt = AllocSetContextCreate(CurrentMemoryContext,
"SPI Plan",
ALLOCSET_SMALL_SIZES);
oldcxt = MemoryContextSwitchTo(plancxt);
/* Copy the SPI plan into its own context */
newplan = (SPIPlanPtr) palloc0(sizeof(_SPI_plan));
newplan->magic = _SPI_PLAN_MAGIC;
newplan->plancxt = plancxt;
newplan->parse_mode = plan->parse_mode;
newplan->cursor_options = plan->cursor_options;
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;
newplan->parserSetup = plan->parserSetup;
newplan->parserSetupArg = plan->parserSetupArg;
/* Copy all the plancache entries */
foreach(lc, plan->plancache_list)
{
CachedPlanSource *plansource = (CachedPlanSource *) lfirst(lc);
CachedPlanSource *newsource;
newsource = CopyCachedPlan(plansource);
newplan->plancache_list = lappend(newplan->plancache_list, newsource);
}
MemoryContextSwitchTo(oldcxt);
/*
* Mark it saved, reparent it under CacheMemoryContext, and mark all the
* component CachedPlanSources as saved. This sequence cannot fail
* partway through, so there's no risk of long-term memory leakage.
*/
newplan->saved = true;
MemoryContextSetParent(newplan->plancxt, CacheMemoryContext);
foreach(lc, newplan->plancache_list)
{
CachedPlanSource *plansource = (CachedPlanSource *) lfirst(lc);
SaveCachedPlan(plansource);
}
return newplan;
}
/*
* Internal lookup of ephemeral named relation by name.
*/
static EphemeralNamedRelation
_SPI_find_ENR_by_name(const char *name)
{
/* internal static function; any error is bug in SPI itself */
Assert(name != NULL);
/* fast exit if no tuplestores have been added */
if (_SPI_current->queryEnv == NULL)
return NULL;
return get_ENR(_SPI_current->queryEnv, name);
}
/*
* Register an ephemeral named relation for use by the planner and executor on
* subsequent calls using this SPI connection.
*/
int
SPI_register_relation(EphemeralNamedRelation enr)
{
EphemeralNamedRelation match;
int res;
if (enr == NULL || enr->md.name == NULL)
return SPI_ERROR_ARGUMENT;
res = _SPI_begin_call(false); /* keep current memory context */
if (res < 0)
return res;
match = _SPI_find_ENR_by_name(enr->md.name);
if (match)
res = SPI_ERROR_REL_DUPLICATE;
else
{
if (_SPI_current->queryEnv == NULL)
_SPI_current->queryEnv = create_queryEnv();
register_ENR(_SPI_current->queryEnv, enr);
res = SPI_OK_REL_REGISTER;
}
_SPI_end_call(false);
return res;
}
/*
* Unregister an ephemeral named relation by name. This will probably be a
* rarely used function, since SPI_finish will clear it automatically.
*/
int
SPI_unregister_relation(const char *name)
{
EphemeralNamedRelation match;
int res;
if (name == NULL)
return SPI_ERROR_ARGUMENT;
res = _SPI_begin_call(false); /* keep current memory context */
if (res < 0)
return res;
match = _SPI_find_ENR_by_name(name);
if (match)
{
unregister_ENR(_SPI_current->queryEnv, match->md.name);
res = SPI_OK_REL_UNREGISTER;
}
else
res = SPI_ERROR_REL_NOT_FOUND;
_SPI_end_call(false);
return res;
}
/*
* Register the transient relations from 'tdata' using this SPI connection.
* This should be called by PL implementations' trigger handlers after
* connecting, in order to make transition tables visible to any queries run
* in this connection.
*/
int
SPI_register_trigger_data(TriggerData *tdata)
{
if (tdata == NULL)
return SPI_ERROR_ARGUMENT;
if (tdata->tg_newtable)
{
EphemeralNamedRelation enr =
palloc(sizeof(EphemeralNamedRelationData));
int rc;
enr->md.name = tdata->tg_trigger->tgnewtable;
enr->md.reliddesc = tdata->tg_relation->rd_id;
enr->md.tupdesc = NULL;
enr->md.enrtype = ENR_NAMED_TUPLESTORE;
enr->md.enrtuples = tuplestore_tuple_count(tdata->tg_newtable);
enr->reldata = tdata->tg_newtable;
rc = SPI_register_relation(enr);
if (rc != SPI_OK_REL_REGISTER)
return rc;
}
if (tdata->tg_oldtable)
{
EphemeralNamedRelation enr =
palloc(sizeof(EphemeralNamedRelationData));
int rc;
enr->md.name = tdata->tg_trigger->tgoldtable;
enr->md.reliddesc = tdata->tg_relation->rd_id;
enr->md.tupdesc = NULL;
enr->md.enrtype = ENR_NAMED_TUPLESTORE;
enr->md.enrtuples = tuplestore_tuple_count(tdata->tg_oldtable);
enr->reldata = tdata->tg_oldtable;
rc = SPI_register_relation(enr);
if (rc != SPI_OK_REL_REGISTER)
return rc;
}
return SPI_OK_TD_REGISTER;
}
Reference in New Issue View Git Blame Copy Permalink