pquery.c

来自「PostgreSQL7.4.6 for Linux」· C语言 代码 · 共 1,183 行 · 第 1/3 页

/*-------------------------------------------------------------------------
 *
 * pquery.c
 *	  POSTGRES process query command code
 *
 * Portions Copyright (c) 1996-2003, PostgreSQL Global Development Group
 * Portions Copyright (c) 1994, Regents of the University of California
 *
 *
 * IDENTIFICATION
 *	  $Header: /cvsroot/pgsql/src/backend/tcop/pquery.c,v 1.73.2.1 2004/03/05 00:21:51 momjian Exp $
 *
 *-------------------------------------------------------------------------
 */
#include "postgres.h"

#include "executor/executor.h"
#include "miscadmin.h"
#include "tcop/tcopprot.h"
#include "tcop/pquery.h"
#include "tcop/utility.h"
#include "utils/guc.h"
#include "utils/memutils.h"


static uint32 RunFromStore(Portal portal, ScanDirection direction, long count,
			 DestReceiver *dest);
static long PortalRunSelect(Portal portal, bool forward, long count,
				DestReceiver *dest);
static void PortalRunUtility(Portal portal, Query *query,
				 DestReceiver *dest, char *completionTag);
static void PortalRunMulti(Portal portal,
			   DestReceiver *dest, DestReceiver *altdest,
			   char *completionTag);
static long DoPortalRunFetch(Portal portal,
				 FetchDirection fdirection,
				 long count,
				 DestReceiver *dest);
static void DoPortalRewind(Portal portal);


/*
 * CreateQueryDesc
 */
QueryDesc *
CreateQueryDesc(Query *parsetree,
				Plan *plantree,
				DestReceiver *dest,
				ParamListInfo params,
				bool doInstrument)
{
	QueryDesc  *qd = (QueryDesc *) palloc(sizeof(QueryDesc));

	qd->operation = parsetree->commandType;		/* operation */
	qd->parsetree = parsetree;	/* parse tree */
	qd->plantree = plantree;	/* plan */
	qd->dest = dest;			/* output dest */
	qd->params = params;		/* parameter values passed into query */
	qd->doInstrument = doInstrument;	/* instrumentation wanted? */

	/* null these fields until set by ExecutorStart */
	qd->tupDesc = NULL;
	qd->estate = NULL;
	qd->planstate = NULL;

	return qd;
}

/*
 * FreeQueryDesc
 */
void
FreeQueryDesc(QueryDesc *qdesc)
{
	/* Can't be a live query */
	Assert(qdesc->estate == NULL);
	/* Only the QueryDesc itself need be freed */
	pfree(qdesc);
}


/*
 * ProcessQuery
 *		Execute a single query
 *
 *	parsetree: the query tree
 *	plan: the plan tree for the query
 *	params: any parameters needed
 *	dest: where to send results
 *	completionTag: points to a buffer of size COMPLETION_TAG_BUFSIZE
 *		in which to store a command completion status string.
 *
 * completionTag may be NULL if caller doesn't want a status string.
 *
 * Must be called in a memory context that will be reset or deleted on
 * error; otherwise the executor's memory usage will be leaked.
 */
void
ProcessQuery(Query *parsetree,
			 Plan *plan,
			 ParamListInfo params,
			 DestReceiver *dest,
			 char *completionTag)
{
	int			operation = parsetree->commandType;
	QueryDesc  *queryDesc;

	/*
	 * Check for special-case destinations
	 */
	if (operation == CMD_SELECT)
	{
		if (parsetree->into != NULL)
		{
			/*
			 * SELECT INTO table (a/k/a CREATE AS ... SELECT).
			 *
			 * Override the normal communication destination; execMain.c
			 * special-cases this case.  (Perhaps would be cleaner to have
			 * an additional destination type?)
			 */
			dest = None_Receiver;
		}
	}

	/*
	 * Create the QueryDesc object
	 */
	queryDesc = CreateQueryDesc(parsetree, plan, dest, params, false);

	/*
	 * Call ExecStart to prepare the plan for execution
	 */
	ExecutorStart(queryDesc, false, false);

	/*
	 * Run the plan to completion.
	 */
	ExecutorRun(queryDesc, ForwardScanDirection, 0L);

	/*
	 * Build command completion status string, if caller wants one.
	 */
	if (completionTag)
	{
		Oid			lastOid;

		switch (operation)
		{
			case CMD_SELECT:
				strcpy(completionTag, "SELECT");
				break;
			case CMD_INSERT:
				if (queryDesc->estate->es_processed == 1)
					lastOid = queryDesc->estate->es_lastoid;
				else
					lastOid = InvalidOid;
				snprintf(completionTag, COMPLETION_TAG_BUFSIZE,
				"INSERT %u %u", lastOid, queryDesc->estate->es_processed);
				break;
			case CMD_UPDATE:
				snprintf(completionTag, COMPLETION_TAG_BUFSIZE,
						 "UPDATE %u", queryDesc->estate->es_processed);
				break;
			case CMD_DELETE:
				snprintf(completionTag, COMPLETION_TAG_BUFSIZE,
						 "DELETE %u", queryDesc->estate->es_processed);
				break;
			default:
				strcpy(completionTag, "???");
				break;
		}
	}

	/*
	 * Now, we close down all the scans and free allocated resources.
	 */
	ExecutorEnd(queryDesc);

	FreeQueryDesc(queryDesc);
}

/*
 * ChoosePortalStrategy
 *		Select portal execution strategy given the intended query list.
 *
 * See the comments in portal.h.
 */
PortalStrategy
ChoosePortalStrategy(List *parseTrees)
{
	PortalStrategy strategy;

	strategy = PORTAL_MULTI_QUERY;		/* default assumption */

	if (length(parseTrees) == 1)
	{
		Query	   *query = (Query *) lfirst(parseTrees);

		if (query->commandType == CMD_SELECT &&
			query->canSetTag &&
			query->into == NULL)
			strategy = PORTAL_ONE_SELECT;
		else if (query->commandType == CMD_UTILITY &&
				 query->canSetTag &&
				 query->utilityStmt != NULL)
		{
			if (UtilityReturnsTuples(query->utilityStmt))
				strategy = PORTAL_UTIL_SELECT;
		}
	}
	return strategy;
}

/*
 * PortalStart
 *		Prepare a portal for execution.
 *
 * Caller must already have created the portal, done PortalDefineQuery(),
 * and adjusted portal options if needed.  If parameters are needed by
 * the query, they must be passed in here (caller is responsible for
 * giving them appropriate lifetime).
 *
 * On return, portal is ready to accept PortalRun() calls, and the result
 * tupdesc (if any) is known.
 */
void
PortalStart(Portal portal, ParamListInfo params)
{
	MemoryContext oldContext;
	QueryDesc  *queryDesc;

	AssertArg(PortalIsValid(portal));
	AssertState(portal->queryContext != NULL);	/* query defined? */
	AssertState(!portal->portalReady);	/* else extra PortalStart */

	oldContext = MemoryContextSwitchTo(PortalGetHeapMemory(portal));

	/* Must remember portal param list, if any */
	portal->portalParams = params;

	/*
	 * Determine the portal execution strategy
	 */
	portal->strategy = ChoosePortalStrategy(portal->parseTrees);

	/*
	 * Fire her up according to the strategy
	 */
	switch (portal->strategy)
	{
		case PORTAL_ONE_SELECT:

			/*
			 * Must set query snapshot before starting executor.
			 */
			SetQuerySnapshot();

			/*
			 * Create QueryDesc in portal's context; for the moment, set
			 * the destination to None.
			 */
			queryDesc = CreateQueryDesc((Query *) lfirst(portal->parseTrees),
									  (Plan *) lfirst(portal->planTrees),
										None_Receiver,
										params,
										false);

			/*
			 * Call ExecStart to prepare the plan for execution
			 */
			ExecutorStart(queryDesc, false, false);

			/*
			 * This tells PortalCleanup to shut down the executor
			 */
			portal->queryDesc = queryDesc;

			/*
			 * Remember tuple descriptor (computed by ExecutorStart)
			 */
			portal->tupDesc = queryDesc->tupDesc;

			/*
			 * Reset cursor position data to "start of query"
			 */
			portal->atStart = true;
			portal->atEnd = false;		/* allow fetches */
			portal->portalPos = 0;
			portal->posOverflow = false;
			break;

		case PORTAL_UTIL_SELECT:

			/*
			 * We don't set query snapshot here, because PortalRunUtility
			 * will take care of it.
			 */
			portal->tupDesc =
				UtilityTupleDescriptor(((Query *) lfirst(portal->parseTrees))->utilityStmt);

			/*
			 * Reset cursor position data to "start of query"
			 */
			portal->atStart = true;
			portal->atEnd = false;		/* allow fetches */
			portal->portalPos = 0;
			portal->posOverflow = false;
			break;

		case PORTAL_MULTI_QUERY:
			/* Need do nothing now */
			portal->tupDesc = NULL;
			break;
	}

	MemoryContextSwitchTo(oldContext);

	portal->portalReady = true;
}

/*
 * PortalSetResultFormat
 *		Select the format codes for a portal's output.
 *
 * This must be run after PortalStart for a portal that will be read by
 * a Remote or RemoteExecute destination.  It is not presently needed for
 * other destination types.
 *
 * formats[] is the client format request, as per Bind message conventions.
 */
void
PortalSetResultFormat(Portal portal, int nFormats, int16 *formats)
{
	int			natts;
	int			i;

	/* Do nothing if portal won't return tuples */
	if (portal->tupDesc == NULL)
		return;
	natts = portal->tupDesc->natts;
	/* +1 avoids palloc(0) if no columns */
	portal->formats = (int16 *)
		MemoryContextAlloc(PortalGetHeapMemory(portal),
						   (natts + 1) * sizeof(int16));
	if (nFormats > 1)
	{
		/* format specified for each column */
		if (nFormats != natts)
			ereport(ERROR,
					(errcode(ERRCODE_PROTOCOL_VIOLATION),
					 errmsg("bind message has %d result formats but query has %d columns",
							nFormats, natts)));
		memcpy(portal->formats, formats, natts * sizeof(int16));
	}
	else if (nFormats > 0)
	{
		/* single format specified, use for all columns */
		int16		format1 = formats[0];

		for (i = 0; i < natts; i++)
			portal->formats[i] = format1;
	}
	else
	{
		/* use default format for all columns */
		for (i = 0; i < natts; i++)
			portal->formats[i] = 0;
	}
}

/*
 * PortalRun
 *		Run a portal's query or queries.
 *
 * count <= 0 is interpreted as a no-op: the destination gets started up
 * and shut down, but nothing else happens.  Also, count == FETCH_ALL is
 * interpreted as "all rows".  Note that count is ignored in multi-query
 * situations, where we always run the portal to completion.
 *
 * dest: where to send output of primary (canSetTag) query
 *
 * altdest: where to send output of non-primary queries
 *
 * completionTag: points to a buffer of size COMPLETION_TAG_BUFSIZE
 *		in which to store a command completion status string.
 *		May be NULL if caller doesn't want a status string.
 *
 * Returns TRUE if the portal's execution is complete, FALSE if it was
 * suspended due to exhaustion of the count parameter.
 */
bool
PortalRun(Portal portal, long count,
		  DestReceiver *dest, DestReceiver *altdest,
		  char *completionTag)