execmain.c

关系型数据库 Postgresql 6.5.2

/*-------------------------------------------------------------------------
 *
 * execMain.c
 *	  top level executor interface routines
 *
 * INTERFACE ROUTINES
 *	ExecutorStart()
 *	ExecutorRun()
 *	ExecutorEnd()
 *
 *	The old ExecutorMain() has been replaced by ExecutorStart(),
 *	ExecutorRun() and ExecutorEnd()
 *
 *	These three procedures are the external interfaces to the executor.
 *	In each case, the query descriptor and the execution state is required
 *	 as arguments
 *
 *	ExecutorStart() must be called at the beginning of any execution of any
 *	query plan and ExecutorEnd() should always be called at the end of
 *	execution of a plan.
 *
 *	ExecutorRun accepts 'feature' and 'count' arguments that specify whether
 *	the plan is to be executed forwards, backwards, and for how many tuples.
 *
 * Copyright (c) 1994, Regents of the University of California
 *
 *
 * IDENTIFICATION
 *	  $Header: /usr/local/cvsroot/pgsql/src/backend/executor/execMain.c,v 1.88 1999/06/17 15:15:49 momjian Exp $
 *
 *-------------------------------------------------------------------------
 */
#include <string.h>
#include "postgres.h"
#include "miscadmin.h"

#include "executor/executor.h"
#include "executor/execdefs.h"
#include "executor/execdebug.h"
#include "executor/nodeIndexscan.h"
#include "utils/builtins.h"
#include "utils/palloc.h"
#include "utils/acl.h"
#include "utils/syscache.h"
#include "utils/tqual.h"
#include "parser/parsetree.h"	/* rt_fetch() */
#include "storage/bufmgr.h"
#include "storage/lmgr.h"
#include "storage/smgr.h"
#include "commands/async.h"
/* #include "access/localam.h" */
#include "optimizer/var.h"
#include "access/heapam.h"
#include "access/xact.h"
#include "catalog/heap.h"
#include "commands/trigger.h"

void ExecCheckPerms(CmdType operation, int resultRelation, List *rangeTable,
			   Query *parseTree);


/* decls for local routines only used within this module */
static TupleDesc InitPlan(CmdType operation,
		 Query *parseTree,
		 Plan *plan,
		 EState *estate);
static void EndPlan(Plan *plan,
		EState *estate);
static TupleTableSlot *ExecutePlan(EState *estate, Plan *plan,
			CmdType operation,
			int offsetTuples,
			int numberTuples,
			ScanDirection direction,
			DestReceiver *destfunc);
static void ExecRetrieve(TupleTableSlot *slot,
			 DestReceiver *destfunc,
			 EState *estate);
static void ExecAppend(TupleTableSlot *slot, ItemPointer tupleid,
		   EState *estate);
static void ExecDelete(TupleTableSlot *slot, ItemPointer tupleid,
		   EState *estate);
static void ExecReplace(TupleTableSlot *slot, ItemPointer tupleid,
			EState *estate);

TupleTableSlot *EvalPlanQual(EState *estate, Index rti, ItemPointer tid);
static TupleTableSlot *EvalPlanQualNext(EState *estate);


/* end of local decls */

/* ----------------------------------------------------------------
 *		ExecutorStart
 *
 *		This routine must be called at the beginning of any execution of any
 *		query plan
 *
 *		returns (AttrInfo*) which describes the attributes of the tuples to
 *		be returned by the query.
 *
 * ----------------------------------------------------------------
 */
TupleDesc
ExecutorStart(QueryDesc *queryDesc, EState *estate)
{
	TupleDesc	result;

	/* sanity checks */
	Assert(queryDesc != NULL);

	if (queryDesc->plantree->nParamExec > 0)
	{
		estate->es_param_exec_vals = (ParamExecData *)
			palloc(queryDesc->plantree->nParamExec * sizeof(ParamExecData));
		memset(estate->es_param_exec_vals, 0, queryDesc->plantree->nParamExec * sizeof(ParamExecData));
	}

	/*
	 * Make our own private copy of the current queries snapshot data
	 */
	if (QuerySnapshot == NULL)
		estate->es_snapshot = NULL;
	else
	{
		estate->es_snapshot = (Snapshot) palloc(sizeof(SnapshotData));
		memcpy(estate->es_snapshot, QuerySnapshot, sizeof(SnapshotData));
		if (estate->es_snapshot->xcnt > 0)
		{
			estate->es_snapshot->xip = (TransactionId *)
				palloc(estate->es_snapshot->xcnt * sizeof(TransactionId));
			memcpy(estate->es_snapshot->xip, QuerySnapshot->xip,
				   estate->es_snapshot->xcnt * sizeof(TransactionId));
		}
	}

	/*
	 * Initialize the plan
	 */
	result = InitPlan(queryDesc->operation,
					  queryDesc->parsetree,
					  queryDesc->plantree,
					  estate);

	/*
	 * reset buffer refcount.  the current refcounts are saved and will be
	 * restored when ExecutorEnd is called
	 *
	 * this makes sure that when ExecutorRun's are called recursively as for
	 * postquel functions, the buffers pinned by one ExecutorRun will not
	 * be unpinned by another ExecutorRun.
	 */
	BufferRefCountReset(estate->es_refcount);

	return result;
}

/* ----------------------------------------------------------------
 *		ExecutorRun
 *
 *		This is the main routine of the executor module. It accepts
 *		the query descriptor from the traffic cop and executes the
 *		query plan.
 *
 *		ExecutorStart must have been called already.
 *
 *		the different features supported are:
 *			 EXEC_RUN:	retrieve all tuples in the forward direction
 *			 EXEC_FOR:	retrieve 'count' number of tuples in the forward dir
 *			 EXEC_BACK: retrieve 'count' number of tuples in the backward dir
 *			 EXEC_RETONE: return one tuple but don't 'retrieve' it
 *						   used in postquel function processing
 *
 *
 * ----------------------------------------------------------------
 */
TupleTableSlot *
ExecutorRun(QueryDesc *queryDesc, EState *estate, int feature,
			Node *limoffset, Node *limcount)
{
	CmdType		operation;
	Plan	   *plan;
	TupleTableSlot *result;
	CommandDest dest;
	DestReceiver *destfunc;
	int			offset = 0;
	int			count = 0;

	/*
	 * sanity checks
	 */
	Assert(queryDesc != NULL);

	/*
	 * extract information from the query descriptor and the query
	 * feature.
	 */
	operation = queryDesc->operation;
	plan = queryDesc->plantree;
	dest = queryDesc->dest;
	destfunc = DestToFunction(dest);
	estate->es_processed = 0;
	estate->es_lastoid = InvalidOid;

	/*
	 * FIXME: the dest setup function ought to be handed the tuple desc
	 * for the tuples to be output, but I'm not quite sure how to get that
	 * info at this point.	For now, passing NULL is OK because no
	 * existing dest setup function actually uses the pointer.
	 */
	(*destfunc->setup) (destfunc, (TupleDesc) NULL);

	/*
	 * if given get the offset of the LIMIT clause
	 */
	if (limoffset != NULL)
	{
		Const	   *coffset;
		Param	   *poffset;
		ParamListInfo paramLI;
		int			i;

		switch (nodeTag(limoffset))
		{
			case T_Const:
				coffset = (Const *) limoffset;
				offset = (int) (coffset->constvalue);
				break;

			case T_Param:
				poffset = (Param *) limoffset;
				paramLI = estate->es_param_list_info;

				if (paramLI == NULL)
					elog(ERROR, "parameter for limit offset not in executor state");
				for (i = 0; paramLI[i].kind != PARAM_INVALID; i++)
				{
					if (paramLI[i].kind == PARAM_NUM && paramLI[i].id == poffset->paramid)
						break;
				}
				if (paramLI[i].kind == PARAM_INVALID)
					elog(ERROR, "parameter for limit offset not in executor state");
				if (paramLI[i].isnull)
					elog(ERROR, "limit offset cannot be NULL value");
				offset = (int) (paramLI[i].value);

				break;

			default:
				elog(ERROR, "unexpected node type %d as limit offset", nodeTag(limoffset));
		}

		if (offset < 0)
			elog(ERROR, "limit offset cannot be negative");
	}

	/*
	 * if given get the count of the LIMIT clause
	 */
	if (limcount != NULL)
	{
		Const	   *ccount;
		Param	   *pcount;
		ParamListInfo paramLI;
		int			i;

		switch (nodeTag(limcount))
		{
			case T_Const:
				ccount = (Const *) limcount;
				count = (int) (ccount->constvalue);
				break;

			case T_Param:
				pcount = (Param *) limcount;
				paramLI = estate->es_param_list_info;

				if (paramLI == NULL)
					elog(ERROR, "parameter for limit count not in executor state");
				for (i = 0; paramLI[i].kind != PARAM_INVALID; i++)
				{
					if (paramLI[i].kind == PARAM_NUM && paramLI[i].id == pcount->paramid)
						break;
				}
				if (paramLI[i].kind == PARAM_INVALID)
					elog(ERROR, "parameter for limit count not in executor state");
				if (paramLI[i].isnull)
					elog(ERROR, "limit count cannot be NULL value");
				count = (int) (paramLI[i].value);

				break;

			default:
				elog(ERROR, "unexpected node type %d as limit count", nodeTag(limcount));
		}

		if (count < 0)
			elog(ERROR, "limit count cannot be negative");
	}

	switch (feature)
	{

		case EXEC_RUN:
			result = ExecutePlan(estate,
								 plan,
								 operation,
								 offset,
								 count,
								 ForwardScanDirection,
								 destfunc);
			break;
		case EXEC_FOR:
			result = ExecutePlan(estate,
								 plan,
								 operation,
								 offset,
								 count,
								 ForwardScanDirection,
								 destfunc);
			break;

			/*
			 * retrieve next n "backward" tuples
			 */
		case EXEC_BACK:
			result = ExecutePlan(estate,
								 plan,
								 operation,
								 offset,
								 count,
								 BackwardScanDirection,
								 destfunc);
			break;

			/*
			 * return one tuple but don't "retrieve" it. (this is used by
			 * the rule manager..) -cim 9/14/89
			 */
		case EXEC_RETONE:
			result = ExecutePlan(estate,
								 plan,
								 operation,
								 0,
								 ONE_TUPLE,
								 ForwardScanDirection,
								 destfunc);
			break;
		default:
			result = NULL;
			elog(DEBUG, "ExecutorRun: Unknown feature %d", feature);
			break;
	}

	(*destfunc->cleanup) (destfunc);

	return result;
}

/* ----------------------------------------------------------------
 *		ExecutorEnd
 *
 *		This routine must be called at the end of any execution of any
 *		query plan
 *
 *		returns (AttrInfo*) which describes the attributes of the tuples to
 *		be returned by the query.
 *
 * ----------------------------------------------------------------
 */
void
ExecutorEnd(QueryDesc *queryDesc, EState *estate)
{
	/* sanity checks */
	Assert(queryDesc != NULL);

	EndPlan(queryDesc->plantree, estate);

	/* XXX - clean up some more from ExecutorStart() - er1p */
	if (NULL == estate->es_snapshot)
	{
		/* nothing to free */
	}
	else
	{
		if (estate->es_snapshot->xcnt > 0)
			pfree(estate->es_snapshot->xip);
		pfree(estate->es_snapshot);
	}

	if (NULL == estate->es_param_exec_vals)
	{
		/* nothing to free */
	}
	else
	{
		pfree(estate->es_param_exec_vals);
		estate->es_param_exec_vals = NULL;
	}

	/* restore saved refcounts. */
	BufferRefCountRestore(estate->es_refcount);

}

void
ExecCheckPerms(CmdType operation,
			   int resultRelation,
			   List *rangeTable,
			   Query *parseTree)
{
	int			i = 1;
	Oid			relid;
	HeapTuple	htup;
	List	   *lp;
	List	   *qvars,
			   *tvars;
	int32		ok = 1,
				aclcheck_result = -1;
	char	   *opstr;
	NameData	rname;
	char	   *userName;

#define CHECK(MODE)		pg_aclcheck(rname.data, userName, MODE)

	userName = GetPgUserName();

	foreach(lp, rangeTable)
	{
		RangeTblEntry *rte = lfirst(lp);

		if (rte->skipAcl)
		{

			/*
			 * This happens if the access to this table is due to a view
			 * query rewriting - the rewrite handler checked the
			 * permissions against the view owner, so we just skip this
			 * entry.
			 */
			continue;
		}

		relid = rte->relid;
		htup = SearchSysCacheTuple(RELOID,
								   ObjectIdGetDatum(relid),
								   0, 0, 0);
		if (!HeapTupleIsValid(htup))
			elog(ERROR, "ExecCheckPerms: bogus RT relid: %u", relid);
		StrNCpy(rname.data,
				((Form_pg_class) GETSTRUCT(htup))->relname.data,
				NAMEDATALEN);
		if (i == resultRelation)
		{						/* this is the result relation */
			qvars = pull_varnos(parseTree->qual);
			tvars = pull_varnos((Node *) parseTree->targetList);
			if (intMember(resultRelation, qvars) ||
				intMember(resultRelation, tvars))
			{
				/* result relation is scanned */
				ok = ((aclcheck_result = CHECK(ACL_RD)) == ACLCHECK_OK);
				opstr = "read";
				if (!ok)
					break;
			}
			switch (operation)
			{
				case CMD_INSERT:
					ok = ((aclcheck_result = CHECK(ACL_AP)) == ACLCHECK_OK) ||
						((aclcheck_result = CHECK(ACL_WR)) == ACLCHECK_OK);
					opstr = "append";
					break;