spi.c

关系型数据库 Postgresql 6.5.2

	MemoryContext oldcxt = NULL;

	if (_SPI_curid + 1 == _SPI_connected)		/* connected */
	{
		if (_SPI_current != &(_SPI_stack[_SPI_curid + 1]))
			elog(FATAL, "SPI: stack corrupted");
		oldcxt = MemoryContextSwitchTo(_SPI_current->savedcxt);
	}

	pointer = repalloc(pointer, size);

	if (oldcxt)
		MemoryContextSwitchTo(oldcxt);

	return pointer;
}

void
SPI_pfree(void *pointer)
{
	MemoryContext oldcxt = NULL;

	if (_SPI_curid + 1 == _SPI_connected)		/* connected */
	{
		if (_SPI_current != &(_SPI_stack[_SPI_curid + 1]))
			elog(FATAL, "SPI: stack corrupted");
		oldcxt = MemoryContextSwitchTo(_SPI_current->savedcxt);
	}

	pfree(pointer);

	if (oldcxt)
		MemoryContextSwitchTo(oldcxt);

	return;
}

/* =================== private functions =================== */

/*
 * spi_printtup
 *		store tuple retrieved by Executor into SPITupleTable
 *		of current SPI procedure
 *
 */
void
spi_printtup(HeapTuple tuple, TupleDesc tupdesc, DestReceiver *self)
{
	SPITupleTable *tuptable;
	MemoryContext oldcxt;

	/*
	 * When called by Executor _SPI_curid expected to be equal to
	 * _SPI_connected
	 */
	if (_SPI_curid != _SPI_connected || _SPI_connected < 0)
		elog(FATAL, "SPI: improper call to spi_printtup");
	if (_SPI_current != &(_SPI_stack[_SPI_curid]))
		elog(FATAL, "SPI: stack corrupted in spi_printtup");

	oldcxt = _SPI_procmem();	/* switch to procedure memory context */

	tuptable = _SPI_current->tuptable;
	if (tuptable == NULL)
	{
		_SPI_current->tuptable = tuptable = (SPITupleTable *)
			palloc(sizeof(SPITupleTable));
		tuptable->alloced = tuptable->free = 128;
		tuptable->vals = (HeapTuple *) palloc(tuptable->alloced * sizeof(HeapTuple));
		tuptable->tupdesc = CreateTupleDescCopy(tupdesc);
	}
	else if (tuptable->free == 0)
	{
		tuptable->free = 256;
		tuptable->alloced += tuptable->free;
		tuptable->vals = (HeapTuple *) repalloc(tuptable->vals,
								  tuptable->alloced * sizeof(HeapTuple));
	}

	tuptable->vals[tuptable->alloced - tuptable->free] = heap_copytuple(tuple);
	(tuptable->free)--;

	MemoryContextSwitchTo(oldcxt);
	return;
}

/*
 * Static functions
 */

static int
_SPI_execute(char *src, int tcount, _SPI_plan *plan)
{
	List	   *queryTree_list;
	List	   *planTree_list;
	List	   *queryTree_list_item;
	List	   *ptlist;
	QueryDesc  *qdesc;
	Query	   *queryTree;
	Plan	   *planTree;
	EState	   *state;
	int			nargs = 0;
	Oid		   *argtypes = NULL;
	int			res = 0;
	bool		islastquery;

	/* Increment CommandCounter to see changes made by now */
	CommandCounterIncrement();

	SPI_processed = 0;
	SPI_tuptable = NULL;
	_SPI_current->tuptable = NULL;
	_SPI_current->qtlist = NULL;

	if (plan)
	{
		nargs = plan->nargs;
		argtypes = plan->argtypes;
	}
	ptlist = planTree_list =
		pg_parse_and_plan(src, argtypes, nargs, &queryTree_list, None, FALSE);

	_SPI_current->qtlist = queryTree_list;

	foreach(queryTree_list_item, queryTree_list)
	{
		queryTree = (Query *) lfirst(queryTree_list_item);
		planTree = lfirst(planTree_list);
		planTree_list = lnext(planTree_list);
		islastquery = (planTree_list == NIL);	/* assume lists are same
												 * len */

		if (queryTree->commandType == CMD_UTILITY)
		{
			if (nodeTag(queryTree->utilityStmt) == T_CopyStmt)
			{
				CopyStmt   *stmt = (CopyStmt *) (queryTree->utilityStmt);

				if (stmt->filename == NULL)
					return SPI_ERROR_COPY;
			}
			else if (nodeTag(queryTree->utilityStmt) == T_ClosePortalStmt ||
					 nodeTag(queryTree->utilityStmt) == T_FetchStmt)
				return SPI_ERROR_CURSOR;
			else if (nodeTag(queryTree->utilityStmt) == T_TransactionStmt)
				return SPI_ERROR_TRANSACTION;
			res = SPI_OK_UTILITY;
			if (plan == NULL)
			{
				ProcessUtility(queryTree->utilityStmt, None);
				if (!islastquery)
					CommandCounterIncrement();
				else
					return res;
			}
			else if (islastquery)
				break;
		}
		else if (plan == NULL)
		{
			qdesc = CreateQueryDesc(queryTree, planTree,
									islastquery ? SPI : None);
			state = CreateExecutorState();
			res = _SPI_pquery(qdesc, state, islastquery ? tcount : 0);
			if (res < 0 || islastquery)
				return res;
			CommandCounterIncrement();
		}
		else
		{
			qdesc = CreateQueryDesc(queryTree, planTree,
									islastquery ? SPI : None);
			res = _SPI_pquery(qdesc, NULL, islastquery ? tcount : 0);
			if (res < 0)
				return res;
			if (islastquery)
				break;
		}
	}

	plan->qtlist = queryTree_list;
	plan->ptlist = ptlist;

	return res;
}

static int
_SPI_execute_plan(_SPI_plan *plan, Datum *Values, char *Nulls, int tcount)
{
	List	   *queryTree_list = plan->qtlist;
	List	   *planTree_list = plan->ptlist;
	List	   *queryTree_list_item;
	QueryDesc  *qdesc;
	Query	   *queryTree;
	Plan	   *planTree;
	EState	   *state;
	int			nargs = plan->nargs;
	int			res = 0;
	bool		islastquery;
	int			k;

	/* Increment CommandCounter to see changes made by now */
	CommandCounterIncrement();

	SPI_processed = 0;
	SPI_tuptable = NULL;
	_SPI_current->tuptable = NULL;
	_SPI_current->qtlist = NULL;

	foreach(queryTree_list_item, queryTree_list)
	{
		queryTree = (Query *) lfirst(queryTree_list_item);
		planTree = lfirst(planTree_list);
		planTree_list = lnext(planTree_list);
		islastquery = (planTree_list == NIL);	/* assume lists are same
												 * len */

		if (queryTree->commandType == CMD_UTILITY)
		{
			ProcessUtility(queryTree->utilityStmt, None);
			if (!islastquery)
				CommandCounterIncrement();
			else
				return SPI_OK_UTILITY;
		}
		else
		{
			qdesc = CreateQueryDesc(queryTree, planTree,
									islastquery ? SPI : None);
			state = CreateExecutorState();
			if (nargs > 0)
			{
				ParamListInfo paramLI = (ParamListInfo) palloc((nargs + 1) *
											  sizeof(ParamListInfoData));

				state->es_param_list_info = paramLI;
				for (k = 0; k < plan->nargs; paramLI++, k++)
				{
					paramLI->kind = PARAM_NUM;
					paramLI->id = k + 1;
					paramLI->isnull = (Nulls && Nulls[k] == 'n');
					paramLI->value = Values[k];
				}
				paramLI->kind = PARAM_INVALID;
			}
			else
				state->es_param_list_info = NULL;
			res = _SPI_pquery(qdesc, state, islastquery ? tcount : 0);
			if (res < 0 || islastquery)
				return res;
			CommandCounterIncrement();
		}
	}

	return res;
}

static int
_SPI_pquery(QueryDesc *queryDesc, EState *state, int tcount)
{
	Query	   *parseTree = queryDesc->parsetree;
	Plan	   *plan = queryDesc->plantree;
	int			operation = queryDesc->operation;
	CommandDest dest = queryDesc->dest;
	TupleDesc	tupdesc;
	bool		isRetrieveIntoPortal = false;
	bool		isRetrieveIntoRelation = false;
	char	   *intoName = NULL;
	int			res;
	Const		tcount_const;
	Node	   *count = NULL;

	switch (operation)
	{
		case CMD_SELECT:
			res = SPI_OK_SELECT;
			if (parseTree->isPortal)
			{
				isRetrieveIntoPortal = true;
				intoName = parseTree->into;
				parseTree->isBinary = false;	/* */

				return SPI_ERROR_CURSOR;

			}
			else if (parseTree->into != NULL)	/* select into table */
			{
				res = SPI_OK_SELINTO;
				isRetrieveIntoRelation = true;
				queryDesc->dest = None; /* */
			}
			break;
		case CMD_INSERT:
			res = SPI_OK_INSERT;
			break;
		case CMD_DELETE:
			res = SPI_OK_DELETE;
			break;
		case CMD_UPDATE:
			res = SPI_OK_UPDATE;
			break;
		default:
			return SPI_ERROR_OPUNKNOWN;
	}

	/* ----------------
	 * Get the query LIMIT tuple count
	 * ----------------
	 */
	if (parseTree->limitCount != NULL)
	{
		/* ----------------
		 * A limit clause in the parsetree overrides the
		 * tcount parameter
		 * ----------------
		 */
		count = parseTree->limitCount;
	}
	else
	{
		/* ----------------
		 * No LIMIT clause in parsetree. Use a local Const node
		 * to put tcount into it
		 * ----------------
		 */
		memset(&tcount_const, 0, sizeof(tcount_const));
		tcount_const.type = T_Const;
		tcount_const.consttype = INT4OID;
		tcount_const.constlen = sizeof(int4);
		tcount_const.constvalue = (Datum) tcount;
		tcount_const.constisnull = FALSE;
		tcount_const.constbyval = TRUE;
		tcount_const.constisset = FALSE;
		tcount_const.constiscast = FALSE;

		count = (Node *) &tcount_const;
	}

	if (state == NULL)			/* plan preparation */
		return res;
#ifdef SPI_EXECUTOR_STATS
	if (ShowExecutorStats)
		ResetUsage();
#endif
	tupdesc = ExecutorStart(queryDesc, state);

	/* Don't work currently */
	if (isRetrieveIntoPortal)
	{
		ProcessPortal(intoName,
					  parseTree,
					  plan,
					  state,
					  tupdesc,
					  None);
		return SPI_OK_CURSOR;
	}

	ExecutorRun(queryDesc, state, EXEC_FOR, parseTree->limitOffset, count);

	_SPI_current->processed = state->es_processed;
	if (operation == CMD_SELECT && queryDesc->dest == SPI)
	{
		if (_SPI_checktuples())
			elog(FATAL, "SPI_select: # of processed tuples check failed");
	}

	ExecutorEnd(queryDesc, state);

#ifdef SPI_EXECUTOR_STATS
	if (ShowExecutorStats)
	{
		fprintf(stderr, "! Executor Stats:\n");
		ShowUsage();
	}
#endif

	if (dest == SPI)
	{
		SPI_processed = _SPI_current->processed;
		SPI_tuptable = _SPI_current->tuptable;
	}
	queryDesc->dest = dest;

	return res;

}

static MemoryContext
_SPI_execmem()
{
	MemoryContext oldcxt;
	PortalHeapMemory phmem;

	phmem = PortalGetHeapMemory(_SPI_current->portal);
	oldcxt = MemoryContextSwitchTo((MemoryContext) phmem);

	return oldcxt;

}

static MemoryContext
_SPI_procmem()
{
	MemoryContext oldcxt;
	PortalVariableMemory pvmem;

	pvmem = PortalGetVariableMemory(_SPI_current->portal);
	oldcxt = MemoryContextSwitchTo((MemoryContext) pvmem);

	return oldcxt;

}

/*
 * _SPI_begin_call
 *
 */
static int
_SPI_begin_call(bool execmem)
{
	if (_SPI_curid + 1 != _SPI_connected)
		return SPI_ERROR_UNCONNECTED;
	_SPI_curid++;
	if (_SPI_current != &(_SPI_stack[_SPI_curid]))
		elog(FATAL, "SPI: stack corrupted");

	if (execmem)				/* switch to the Executor memory context */
	{
		_SPI_execmem();
		StartPortalAllocMode(DefaultAllocMode, 0);
	}

	return 0;
}

static int
_SPI_end_call(bool procmem)
{

	/*
	 * We' returning to procedure where _SPI_curid == _SPI_connected - 1
	 */
	_SPI_curid--;

	_SPI_current->qtlist = NULL;

	if (procmem)				/* switch to the procedure memory context */
	{							/* but free Executor memory before */
		EndPortalAllocMode();
		_SPI_procmem();
	}

	return 0;
}

static bool
_SPI_checktuples()
{
	uint32		processed = _SPI_current->processed;
	SPITupleTable *tuptable = _SPI_current->tuptable;
	bool		failed = false;

	if (processed == 0)
	{
		if (tuptable != NULL)
			failed = true;
	}
	else
/* some tuples were processed */
	{
		if (tuptable == NULL)	/* spi_printtup was not called */
			failed = true;
		else if (processed != (tuptable->alloced - tuptable->free))
			failed = true;
	}

	return failed;
}

static _SPI_plan *
_SPI_copy_plan(_SPI_plan *plan, int location)
{
	_SPI_plan  *newplan;
	MemoryContext oldcxt = NULL;

	if (location == _SPI_CPLAN_PROCXT)
		oldcxt = MemoryContextSwitchTo((MemoryContext)
						  PortalGetVariableMemory(_SPI_current->portal));
	else if (location == _SPI_CPLAN_TOPCXT)
		oldcxt = MemoryContextSwitchTo(TopMemoryContext);

	newplan = (_SPI_plan *) palloc(sizeof(_SPI_plan));
	newplan->qtlist = (List *) copyObject(plan->qtlist);
	newplan->ptlist = (List *) copyObject(plan->ptlist);
	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;

	if (location != _SPI_CPLAN_CURCXT)
		MemoryContextSwitchTo(oldcxt);

	return newplan;
}