Header And Logo
|
#include "postgres.h"#include <limits.h>#include "access/xact.h"#include "commands/portalcmds.h"#include "executor/executor.h"#include "executor/tstoreReceiver.h"#include "tcop/pquery.h"#include "utils/memutils.h"#include "utils/snapmgr.h"
Go to the source code of this file.
Functions | |
| void | PerformCursorOpen (PlannedStmt *stmt, ParamListInfo params, const char *queryString, bool isTopLevel) |
| void | PerformPortalFetch (FetchStmt *stmt, DestReceiver *dest, char *completionTag) |
| void | PerformPortalClose (const char *name) |
| void | PortalCleanup (Portal portal) |
| void | PersistHoldablePortal (Portal portal) |
| void PerformCursorOpen | ( | PlannedStmt * | stmt, | |
| ParamListInfo | params, | |||
| const char * | queryString, | |||
| bool | isTopLevel | |||
| ) |
Definition at line 44 of file portalcmds.c.
References Assert, copyObject(), copyParamList(), CreatePortal(), CURSOR_OPT_HOLD, CURSOR_OPT_NO_SCROLL, CURSOR_OPT_SCROLL, elog, ereport, errcode(), errmsg(), ERROR, ExecSupportsBackwardScan(), GetActiveSnapshot(), IsA, list_make1, MemoryContextSwitchTo(), NIL, NULL, DeclareCursorStmt::options, PlannedStmt::planTree, PORTAL_ONE_SELECT, PortalDefineQuery(), PortalGetHeapMemory, DeclareCursorStmt::portalname, PortalStart(), pstrdup(), RequireTransactionChain(), PlannedStmt::rowMarks, and PlannedStmt::utilityStmt.
Referenced by standard_ProcessUtility().
{
DeclareCursorStmt *cstmt = (DeclareCursorStmt *) stmt->utilityStmt;
Portal portal;
MemoryContext oldContext;
if (cstmt == NULL || !IsA(cstmt, DeclareCursorStmt))
elog(ERROR, "PerformCursorOpen called for non-cursor query");
/*
* Disallow empty-string cursor name (conflicts with protocol-level
* unnamed portal).
*/
if (!cstmt->portalname || cstmt->portalname[0] == '\0')
ereport(ERROR,
(errcode(ERRCODE_INVALID_CURSOR_NAME),
errmsg("invalid cursor name: must not be empty")));
/*
* If this is a non-holdable cursor, we require that this statement has
* been executed inside a transaction block (or else, it would have no
* user-visible effect).
*/
if (!(cstmt->options & CURSOR_OPT_HOLD))
RequireTransactionChain(isTopLevel, "DECLARE CURSOR");
/*
* Create a portal and copy the plan and queryString into its memory.
*/
portal = CreatePortal(cstmt->portalname, false, false);
oldContext = MemoryContextSwitchTo(PortalGetHeapMemory(portal));
stmt = copyObject(stmt);
stmt->utilityStmt = NULL; /* make it look like plain SELECT */
queryString = pstrdup(queryString);
PortalDefineQuery(portal,
NULL,
queryString,
"SELECT", /* cursor's query is always a SELECT */
list_make1(stmt),
NULL);
/*----------
* Also copy the outer portal's parameter list into the inner portal's
* memory context. We want to pass down the parameter values in case we
* had a command like
* DECLARE c CURSOR FOR SELECT ... WHERE foo = $1
* This will have been parsed using the outer parameter set and the
* parameter value needs to be preserved for use when the cursor is
* executed.
*----------
*/
params = copyParamList(params);
MemoryContextSwitchTo(oldContext);
/*
* Set up options for portal.
*
* If the user didn't specify a SCROLL type, allow or disallow scrolling
* based on whether it would require any additional runtime overhead to do
* so. Also, we disallow scrolling for FOR UPDATE cursors.
*/
portal->cursorOptions = cstmt->options;
if (!(portal->cursorOptions & (CURSOR_OPT_SCROLL | CURSOR_OPT_NO_SCROLL)))
{
if (stmt->rowMarks == NIL &&
ExecSupportsBackwardScan(stmt->planTree))
portal->cursorOptions |= CURSOR_OPT_SCROLL;
else
portal->cursorOptions |= CURSOR_OPT_NO_SCROLL;
}
/*
* Start execution, inserting parameters if any.
*/
PortalStart(portal, params, 0, GetActiveSnapshot());
Assert(portal->strategy == PORTAL_ONE_SELECT);
/*
* We're done; the query won't actually be run until PerformPortalFetch is
* called.
*/
}
| void PerformPortalClose | ( | const char * | name | ) |
Definition at line 194 of file portalcmds.c.
References ereport, errcode(), errmsg(), ERROR, GetPortalByName(), NULL, PortalDrop(), PortalHashTableDeleteAll(), and PortalIsValid.
Referenced by standard_ProcessUtility().
{
Portal portal;
/* NULL means CLOSE ALL */
if (name == NULL)
{
PortalHashTableDeleteAll();
return;
}
/*
* Disallow empty-string cursor name (conflicts with protocol-level
* unnamed portal).
*/
if (name[0] == '\0')
ereport(ERROR,
(errcode(ERRCODE_INVALID_CURSOR_NAME),
errmsg("invalid cursor name: must not be empty")));
/*
* get the portal from the portal name
*/
portal = GetPortalByName(name);
if (!PortalIsValid(portal))
{
ereport(ERROR,
(errcode(ERRCODE_UNDEFINED_CURSOR),
errmsg("cursor \"%s\" does not exist", name)));
return; /* keep compiler happy */
}
/*
* Note: PortalCleanup is called as a side-effect, if not already done.
*/
PortalDrop(portal, false);
}
| void PerformPortalFetch | ( | FetchStmt * | stmt, | |
| DestReceiver * | dest, | |||
| char * | completionTag | |||
| ) |
Definition at line 146 of file portalcmds.c.
References COMPLETION_TAG_BUFSIZE, FetchStmt::direction, ereport, errcode(), errmsg(), ERROR, GetPortalByName(), FetchStmt::howMany, FetchStmt::ismove, None_Receiver, PortalIsValid, FetchStmt::portalname, PortalRunFetch(), and snprintf().
Referenced by standard_ProcessUtility().
{
Portal portal;
long nprocessed;
/*
* Disallow empty-string cursor name (conflicts with protocol-level
* unnamed portal).
*/
if (!stmt->portalname || stmt->portalname[0] == '\0')
ereport(ERROR,
(errcode(ERRCODE_INVALID_CURSOR_NAME),
errmsg("invalid cursor name: must not be empty")));
/* get the portal from the portal name */
portal = GetPortalByName(stmt->portalname);
if (!PortalIsValid(portal))
{
ereport(ERROR,
(errcode(ERRCODE_UNDEFINED_CURSOR),
errmsg("cursor \"%s\" does not exist", stmt->portalname)));
return; /* keep compiler happy */
}
/* Adjust dest if needed. MOVE wants destination DestNone */
if (stmt->ismove)
dest = None_Receiver;
/* Do it */
nprocessed = PortalRunFetch(portal,
stmt->direction,
stmt->howMany,
dest);
/* Return command status if wanted */
if (completionTag)
snprintf(completionTag, COMPLETION_TAG_BUFSIZE, "%s %ld",
stmt->ismove ? "MOVE" : "FETCH",
nprocessed);
}
| void PersistHoldablePortal | ( | Portal | portal | ) |
Definition at line 303 of file portalcmds.c.
References ActivePortal, Assert, PortalData::atEnd, CreateDestReceiver(), PortalData::createSubid, CreateTupleDescCopy(), CurrentResourceOwner, QueryDesc::dest, DestTuplestore, elog, ereport, errcode(), errmsg(), ERROR, ExecutorEnd(), ExecutorFinish(), ExecutorRewind(), ExecutorRun(), ForwardScanDirection, FreeQueryDesc(), PortalData::holdContext, PortalData::holdStore, InvalidSubTransactionId, MarkPortalFailed(), MemoryContextDeleteChildren(), MemoryContextSwitchTo(), PortalData::name, NULL, PG_CATCH, PG_END_TRY, PG_RE_THROW, PG_TRY, PopActiveSnapshot(), PORTAL_READY, PortalContext, PortalGetHeapMemory, PortalGetQueryDesc, PortalData::portalPos, PortalData::posOverflow, PushActiveSnapshot(), PortalData::queryDesc, _DestReceiver::rDestroy, PortalData::resowner, SetTuplestoreDestReceiverParams(), QueryDesc::snapshot, PortalData::status, PortalData::tupDesc, tuplestore_advance(), and tuplestore_rescan().
Referenced by PreCommit_Portals().
{
QueryDesc *queryDesc = PortalGetQueryDesc(portal);
Portal saveActivePortal;
ResourceOwner saveResourceOwner;
MemoryContext savePortalContext;
MemoryContext oldcxt;
/*
* If we're preserving a holdable portal, we had better be inside the
* transaction that originally created it.
*/
Assert(portal->createSubid != InvalidSubTransactionId);
Assert(queryDesc != NULL);
/*
* Caller must have created the tuplestore already.
*/
Assert(portal->holdContext != NULL);
Assert(portal->holdStore != NULL);
/*
* Before closing down the executor, we must copy the tupdesc into
* long-term memory, since it was created in executor memory.
*/
oldcxt = MemoryContextSwitchTo(portal->holdContext);
portal->tupDesc = CreateTupleDescCopy(portal->tupDesc);
MemoryContextSwitchTo(oldcxt);
/*
* Check for improper portal use, and mark portal active.
*/
if (portal->status != PORTAL_READY)
ereport(ERROR,
(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
errmsg("portal \"%s\" cannot be run", portal->name)));
portal->status = PORTAL_ACTIVE;
/*
* Set up global portal context pointers.
*/
saveActivePortal = ActivePortal;
saveResourceOwner = CurrentResourceOwner;
savePortalContext = PortalContext;
PG_TRY();
{
ActivePortal = portal;
CurrentResourceOwner = portal->resowner;
PortalContext = PortalGetHeapMemory(portal);
MemoryContextSwitchTo(PortalContext);
PushActiveSnapshot(queryDesc->snapshot);
/*
* Rewind the executor: we need to store the entire result set in the
* tuplestore, so that subsequent backward FETCHs can be processed.
*/
ExecutorRewind(queryDesc);
/*
* Change the destination to output to the tuplestore. Note we tell
* the tuplestore receiver to detoast all data passed through it.
*/
queryDesc->dest = CreateDestReceiver(DestTuplestore);
SetTuplestoreDestReceiverParams(queryDesc->dest,
portal->holdStore,
portal->holdContext,
true);
/* Fetch the result set into the tuplestore */
ExecutorRun(queryDesc, ForwardScanDirection, 0L);
(*queryDesc->dest->rDestroy) (queryDesc->dest);
queryDesc->dest = NULL;
/*
* Now shut down the inner executor.
*/
portal->queryDesc = NULL; /* prevent double shutdown */
ExecutorFinish(queryDesc);
ExecutorEnd(queryDesc);
FreeQueryDesc(queryDesc);
/*
* Set the position in the result set: ideally, this could be
* implemented by just skipping straight to the tuple # that we need
* to be at, but the tuplestore API doesn't support that. So we start
* at the beginning of the tuplestore and iterate through it until we
* reach where we need to be. FIXME someday? (Fortunately, the
* typical case is that we're supposed to be at or near the start of
* the result set, so this isn't as bad as it sounds.)
*/
MemoryContextSwitchTo(portal->holdContext);
if (portal->atEnd)
{
/* we can handle this case even if posOverflow */
while (tuplestore_advance(portal->holdStore, true))
/* continue */ ;
}
else
{
long store_pos;
if (portal->posOverflow) /* oops, cannot trust portalPos */
ereport(ERROR,
(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
errmsg("could not reposition held cursor")));
tuplestore_rescan(portal->holdStore);
for (store_pos = 0; store_pos < portal->portalPos; store_pos++)
{
if (!tuplestore_advance(portal->holdStore, true))
elog(ERROR, "unexpected end of tuple stream");
}
}
}
PG_CATCH();
{
/* Uncaught error while executing portal: mark it dead */
MarkPortalFailed(portal);
/* Restore global vars and propagate error */
ActivePortal = saveActivePortal;
CurrentResourceOwner = saveResourceOwner;
PortalContext = savePortalContext;
PG_RE_THROW();
}
PG_END_TRY();
MemoryContextSwitchTo(oldcxt);
/* Mark portal not active */
portal->status = PORTAL_READY;
ActivePortal = saveActivePortal;
CurrentResourceOwner = saveResourceOwner;
PortalContext = savePortalContext;
PopActiveSnapshot();
/*
* We can now release any subsidiary memory of the portal's heap context;
* we'll never use it again. The executor already dropped its context,
* but this will clean up anything that glommed onto the portal's heap via
* PortalContext.
*/
MemoryContextDeleteChildren(PortalGetHeapMemory(portal));
}
| void PortalCleanup | ( | Portal | portal | ) |
Definition at line 243 of file portalcmds.c.
References AssertArg, PortalData::cleanup, CurrentResourceOwner, ExecutorEnd(), ExecutorFinish(), FreeQueryDesc(), PG_CATCH, PG_END_TRY, PG_RE_THROW, PG_TRY, PORTAL_FAILED, PortalCleanup(), PortalGetQueryDesc, PortalIsValid, PortalData::queryDesc, PortalData::resowner, and PortalData::status.
Referenced by PortalCleanup().
{
QueryDesc *queryDesc;
/*
* sanity checks
*/
AssertArg(PortalIsValid(portal));
AssertArg(portal->cleanup == PortalCleanup);
/*
* Shut down executor, if still running. We skip this during error abort,
* since other mechanisms will take care of releasing executor resources,
* and we can't be sure that ExecutorEnd itself wouldn't fail.
*/
queryDesc = PortalGetQueryDesc(portal);
if (queryDesc)
{
/*
* Reset the queryDesc before anything else. This prevents us from
* trying to shut down the executor twice, in case of an error below.
* The transaction abort mechanisms will take care of resource cleanup
* in such a case.
*/
portal->queryDesc = NULL;
if (portal->status != PORTAL_FAILED)
{
ResourceOwner saveResourceOwner;
/* We must make the portal's resource owner current */
saveResourceOwner = CurrentResourceOwner;
PG_TRY();
{
CurrentResourceOwner = portal->resowner;
ExecutorFinish(queryDesc);
ExecutorEnd(queryDesc);
FreeQueryDesc(queryDesc);
}
PG_CATCH();
{
/* Ensure CurrentResourceOwner is restored on error */
CurrentResourceOwner = saveResourceOwner;
PG_RE_THROW();
}
PG_END_TRY();
CurrentResourceOwner = saveResourceOwner;
}
}
}
1.7.1