execmain.c

关系型数据库 Postgresql 6.5.2

				case CMD_DELETE:
				case CMD_UPDATE:
					ok = ((aclcheck_result = CHECK(ACL_WR)) == ACLCHECK_OK);
					opstr = "write";
					break;
				default:
					elog(ERROR, "ExecCheckPerms: bogus operation %d",
						 operation);
			}
		}
		else
		{
			ok = ((aclcheck_result = CHECK(ACL_RD)) == ACLCHECK_OK);
			opstr = "read";
		}
		if (!ok)
			break;
		++i;
	}
	if (!ok)
		elog(ERROR, "%s: %s", rname.data, aclcheck_error_strings[aclcheck_result]);

	if (parseTree != NULL && parseTree->rowMark != NULL)
	{
		foreach(lp, parseTree->rowMark)
		{
			RowMark    *rm = lfirst(lp);

			if (!(rm->info & ROW_ACL_FOR_UPDATE))
				continue;

			relid = ((RangeTblEntry *) nth(rm->rti - 1, rangeTable))->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);
			ok = ((aclcheck_result = CHECK(ACL_WR)) == ACLCHECK_OK);
			opstr = "write";
			if (!ok)
				elog(ERROR, "%s: %s", rname.data, aclcheck_error_strings[aclcheck_result]);
		}
	}
}

/* ===============================================================
 * ===============================================================
						 static routines follow
 * ===============================================================
 * ===============================================================
 */

typedef struct execRowMark
{
	Relation	relation;
	Index		rti;
	char		resname[32];
} execRowMark;

typedef struct evalPlanQual
{
	Plan	   *plan;
	Index		rti;
	EState		estate;
	struct evalPlanQual *free;
} evalPlanQual;

/* ----------------------------------------------------------------
 *		InitPlan
 *
 *		Initializes the query plan: open files, allocate storage
 *		and start up the rule manager
 * ----------------------------------------------------------------
 */
static TupleDesc
InitPlan(CmdType operation, Query *parseTree, Plan *plan, EState *estate)
{
	List	   *rangeTable;
	int			resultRelation;
	Relation	intoRelationDesc;
	TupleDesc	tupType;
	List	   *targetList;
	int			len;

	/*
	 * get information from query descriptor
	 */
	rangeTable = parseTree->rtable;
	resultRelation = parseTree->resultRelation;

#ifndef NO_SECURITY
	ExecCheckPerms(operation, resultRelation, rangeTable, parseTree);
#endif

	/*
	 * initialize the node's execution state
	 */
	estate->es_range_table = rangeTable;

	/*
	 * initialize the BaseId counter so node base_id's are assigned
	 * correctly.  Someday baseid's will have to be stored someplace other
	 * than estate because they should be unique per query planned.
	 */
	estate->es_BaseId = 1;

	/*
	 * initialize result relation stuff
	 */

	if (resultRelation != 0 && operation != CMD_SELECT)
	{

		/*
		 * if we have a result relation, open it and initialize the result
		 * relation info stuff.
		 */
		RelationInfo *resultRelationInfo;
		Index		resultRelationIndex;
		RangeTblEntry *rtentry;
		Oid			resultRelationOid;
		Relation	resultRelationDesc;

		resultRelationIndex = resultRelation;
		rtentry = rt_fetch(resultRelationIndex, rangeTable);
		resultRelationOid = rtentry->relid;
		resultRelationDesc = heap_open(resultRelationOid);

		if (resultRelationDesc->rd_rel->relkind == RELKIND_SEQUENCE)
			elog(ERROR, "You can't change sequence relation %s",
				 resultRelationDesc->rd_rel->relname.data);

		LockRelation(resultRelationDesc, RowExclusiveLock);

		resultRelationInfo = makeNode(RelationInfo);
		resultRelationInfo->ri_RangeTableIndex = resultRelationIndex;
		resultRelationInfo->ri_RelationDesc = resultRelationDesc;
		resultRelationInfo->ri_NumIndices = 0;
		resultRelationInfo->ri_IndexRelationDescs = NULL;
		resultRelationInfo->ri_IndexRelationInfo = NULL;

		/*
		 * open indices on result relation and save descriptors in the
		 * result relation information..
		 */
		if (operation != CMD_DELETE)
			ExecOpenIndices(resultRelationOid, resultRelationInfo);

		estate->es_result_relation_info = resultRelationInfo;
	}
	else
	{

		/*
		 * if no result relation, then set state appropriately
		 */
		estate->es_result_relation_info = NULL;
	}

	/*
	 * Have to lock relations selected for update
	 */
	estate->es_rowMark = NULL;
	if (parseTree->rowMark != NULL)
	{
		Relation	relation;
		Oid			relid;
		RowMark    *rm;
		List	   *l;
		execRowMark *erm;

		foreach(l, parseTree->rowMark)
		{
			rm = lfirst(l);
			relid = ((RangeTblEntry *) nth(rm->rti - 1, rangeTable))->relid;
			relation = heap_open(relid);
			LockRelation(relation, RowShareLock);
			if (!(rm->info & ROW_MARK_FOR_UPDATE))
				continue;
			erm = (execRowMark *) palloc(sizeof(execRowMark));
			erm->relation = relation;
			erm->rti = rm->rti;
			sprintf(erm->resname, "ctid%u", rm->rti);
			estate->es_rowMark = lappend(estate->es_rowMark, erm);
		}
	}

	/*
	 * initialize the executor "tuple" table.
	 */
	{
		int			nSlots = ExecCountSlotsNode(plan);
		TupleTable	tupleTable = ExecCreateTupleTable(nSlots + 10);		/* why add ten? - jolly */

		estate->es_tupleTable = tupleTable;
	}

	/*
	 * initialize the private state information for all the nodes in the
	 * query tree.	This opens files, allocates storage and leaves us
	 * ready to start processing tuples..
	 */
	ExecInitNode(plan, estate, NULL);

	/*
	 * get the tuple descriptor describing the type of tuples to return..
	 * (this is especially important if we are creating a relation with
	 * "retrieve into")
	 */
	tupType = ExecGetTupType(plan);		/* tuple descriptor */
	targetList = plan->targetlist;
	len = ExecTargetListLength(targetList);		/* number of attributes */

	/*
	 * now that we have the target list, initialize the junk filter if
	 * this is a REPLACE or a DELETE query. We also init the junk filter
	 * if this is an append query (there might be some rule lock info
	 * there...) NOTE: in the future we might want to initialize the junk
	 * filter for all queries. SELECT added by [e-mail]
	 * 5/20/98 to allow ORDER/GROUP BY have an identifier missing from the
	 * target.
	 */
	{
		bool		junk_filter_needed = false;
		List	   *tlist;

		if (operation == CMD_SELECT)
		{
			foreach(tlist, targetList)
			{
				TargetEntry *tle = lfirst(tlist);

				if (tle->resdom->resjunk)
				{
					junk_filter_needed = true;
					break;
				}
			}
		}

		if (operation == CMD_UPDATE || operation == CMD_DELETE ||
			operation == CMD_INSERT ||
			(operation == CMD_SELECT && junk_filter_needed))
		{
			JunkFilter *j = (JunkFilter *) ExecInitJunkFilter(targetList);

			estate->es_junkFilter = j;

			if (operation == CMD_SELECT)
				tupType = j->jf_cleanTupType;
		}
		else
			estate->es_junkFilter = NULL;
	}

	/*
	 * initialize the "into" relation
	 */
	intoRelationDesc = (Relation) NULL;

	if (operation == CMD_SELECT)
	{
		char	   *intoName;
		Oid			intoRelationId;
		TupleDesc	tupdesc;

		if (!parseTree->isPortal)
		{

			/*
			 * a select into table
			 */
			if (parseTree->into != NULL)
			{

				/*
				 * create the "into" relation
				 */
				intoName = parseTree->into;

				/*
				 * have to copy tupType to get rid of constraints
				 */
				tupdesc = CreateTupleDescCopy(tupType);

				intoRelationId = heap_create_with_catalog(intoName,
						   tupdesc, RELKIND_RELATION, parseTree->isTemp);

				FreeTupleDesc(tupdesc);

				/*
				 * XXX rather than having to call setheapoverride(true)
				 * and then back to false, we should change the arguments
				 * to heap_open() instead..
				 */
				setheapoverride(true);

				intoRelationDesc = heap_open(intoRelationId);

				setheapoverride(false);
			}
		}
	}

	estate->es_into_relation_descriptor = intoRelationDesc;

	estate->es_origPlan = plan;
	estate->es_evalPlanQual = NULL;
	estate->es_evTuple = NULL;
	estate->es_useEvalPlan = false;

	return tupType;
}

/* ----------------------------------------------------------------
 *		EndPlan
 *
 *		Cleans up the query plan -- closes files and free up storages
 * ----------------------------------------------------------------
 */
static void
EndPlan(Plan *plan, EState *estate)
{
	RelationInfo *resultRelationInfo;
	Relation	intoRelationDesc;

	/*
	 * get information from state
	 */
	resultRelationInfo = estate->es_result_relation_info;
	intoRelationDesc = estate->es_into_relation_descriptor;

	/*
	 * shut down the query
	 */
	ExecEndNode(plan, plan);

	/*
	 * destroy the executor "tuple" table.
	 */
	{
		TupleTable	tupleTable = (TupleTable) estate->es_tupleTable;

		ExecDestroyTupleTable(tupleTable, true);		/* was missing last arg */
		estate->es_tupleTable = NULL;
	}

	/*
	 * close the result relations if necessary
	 */
	if (resultRelationInfo != NULL)
	{
		Relation	resultRelationDesc;

		resultRelationDesc = resultRelationInfo->ri_RelationDesc;
		heap_close(resultRelationDesc);

		/*
		 * close indices on the result relation
		 */
		ExecCloseIndices(resultRelationInfo);
	}

	/*
	 * close the "into" relation if necessary
	 */
	if (intoRelationDesc != NULL)
		heap_close(intoRelationDesc);
}

/* ----------------------------------------------------------------
 *		ExecutePlan
 *
 *		processes the query plan to retrieve 'tupleCount' tuples in the
 *		direction specified.
 *		Retrieves all tuples if tupleCount is 0
 *
 *		result is either a slot containing a tuple in the case
 *		of a RETRIEVE or NULL otherwise.
 *
 * ----------------------------------------------------------------
 */

/* the ctid attribute is a 'junk' attribute that is removed before the
   user can see it*/

static TupleTableSlot *
ExecutePlan(EState *estate,
			Plan *plan,
			CmdType operation,
			int offsetTuples,
			int numberTuples,
			ScanDirection direction,
			DestReceiver *destfunc)
{
	JunkFilter *junkfilter;
	TupleTableSlot *slot;
	ItemPointer tupleid = NULL;
	ItemPointerData tuple_ctid;
	int			current_tuple_count;
	TupleTableSlot *result;

	/*
	 * initialize local variables
	 */
	slot = NULL;
	current_tuple_count = 0;
	result = NULL;

	/*
	 * Set the direction.
	 */
	estate->es_direction = direction;

	/*
	 * Loop until we've processed the proper number of tuples from the
	 * plan..
	 */

	for (;;)
	{

		/*
		 * Execute the plan and obtain a tuple
		 */
		/* at the top level, the parent of a plan (2nd arg) is itself */
lnext:	;
		if (estate->es_useEvalPlan)
		{
			slot = EvalPlanQualNext(estate);
			if (TupIsNull(slot))
				slot = ExecProcNode(plan, plan);
		}
		else
			slot = ExecProcNode(plan, plan);

		/*
		 * if the tuple is null, then we assume there is nothing more to
		 * process so we just return null...
		 */
		if (TupIsNull(slot))
		{
			result = NULL;
			break;
		}

		/*
		 * For now we completely execute the plan and skip result tuples
		 * if requested by LIMIT offset. Finally we should try to do it in
		 * deeper levels if possible (during index scan) - Jan
		 */
		if (offsetTuples > 0)
		{
			--offsetTuples;
			continue;
		}

		/*
		 * if we have a junk filter, then project a new tuple with the
		 * junk removed.
		 *
		 * Store this new "clean" tuple in the place of the original tuple.
		 *
		 * Also, extract all the junk information we need.
		 */
		if ((junkfilter = estate->es_junkFilter) != (JunkFilter *) NULL)
		{