pl_exec.c

来自「postgresql8.3.4源码,开源数据库」· C语言 代码 · 共 2,523 行 · 第 1/5 页

		exec_init_tuple_store(estate);

	/* rettupdesc will be filled by exec_init_tuple_store */
	tupdesc = estate->rettupdesc;
	natts = tupdesc->natts;

	if (stmt->retvarno >= 0)
	{
		PLpgSQL_datum *retvar = estate->datums[stmt->retvarno];

		switch (retvar->dtype)
		{
			case PLPGSQL_DTYPE_VAR:
				{
					PLpgSQL_var *var = (PLpgSQL_var *) retvar;
					Datum		retval = var->value;
					bool		isNull = var->isnull;

					if (natts != 1)
						ereport(ERROR,
								(errcode(ERRCODE_DATATYPE_MISMATCH),
						errmsg("wrong result type supplied in RETURN NEXT")));

					/* coerce type if needed */
					retval = exec_simple_cast_value(retval,
													var->datatype->typoid,
												 tupdesc->attrs[0]->atttypid,
												tupdesc->attrs[0]->atttypmod,
													isNull);

					tuple = heap_form_tuple(tupdesc, &retval, &isNull);

					free_tuple = true;
				}
				break;

			case PLPGSQL_DTYPE_REC:
				{
					PLpgSQL_rec *rec = (PLpgSQL_rec *) retvar;

					if (!HeapTupleIsValid(rec->tup))
						ereport(ERROR,
						  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
						   errmsg("record \"%s\" is not assigned yet",
								  rec->refname),
						   errdetail("The tuple structure of a not-yet-assigned record is indeterminate.")));
					if (!compatible_tupdesc(tupdesc, rec->tupdesc))
						ereport(ERROR,
								(errcode(ERRCODE_DATATYPE_MISMATCH),
						errmsg("wrong record type supplied in RETURN NEXT")));
					tuple = rec->tup;
				}
				break;

			case PLPGSQL_DTYPE_ROW:
				{
					PLpgSQL_row *row = (PLpgSQL_row *) retvar;

					tuple = make_tuple_from_row(estate, row, tupdesc);
					if (tuple == NULL)
						ereport(ERROR,
								(errcode(ERRCODE_DATATYPE_MISMATCH),
						errmsg("wrong record type supplied in RETURN NEXT")));
					free_tuple = true;
				}
				break;

			default:
				elog(ERROR, "unrecognized dtype: %d", retvar->dtype);
				tuple = NULL;	/* keep compiler quiet */
				break;
		}
	}
	else if (stmt->expr)
	{
		Datum		retval;
		bool		isNull;
		Oid			rettype;

		if (natts != 1)
			ereport(ERROR,
					(errcode(ERRCODE_DATATYPE_MISMATCH),
					 errmsg("wrong result type supplied in RETURN NEXT")));

		retval = exec_eval_expr(estate,
								stmt->expr,
								&isNull,
								&rettype);

		/* coerce type if needed */
		retval = exec_simple_cast_value(retval,
										rettype,
										tupdesc->attrs[0]->atttypid,
										tupdesc->attrs[0]->atttypmod,
										isNull);

		tuple = heap_form_tuple(tupdesc, &retval, &isNull);

		free_tuple = true;

		exec_eval_cleanup(estate);
	}
	else
	{
		ereport(ERROR,
				(errcode(ERRCODE_SYNTAX_ERROR),
				 errmsg("RETURN NEXT must have a parameter")));
		tuple = NULL;			/* keep compiler quiet */
	}

	if (HeapTupleIsValid(tuple))
	{
		MemoryContext oldcxt;

		oldcxt = MemoryContextSwitchTo(estate->tuple_store_cxt);
		tuplestore_puttuple(estate->tuple_store, tuple);
		MemoryContextSwitchTo(oldcxt);

		if (free_tuple)
			heap_freetuple(tuple);
	}

	return PLPGSQL_RC_OK;
}

/* ----------
 * exec_stmt_return_query		Evaluate a query and add it to the
 *								list of tuples returned by the current
 *								SRF.
 * ----------
 */
static int
exec_stmt_return_query(PLpgSQL_execstate *estate,
					   PLpgSQL_stmt_return_query *stmt)
{
	Portal		portal;

	if (!estate->retisset)
		ereport(ERROR,
				(errcode(ERRCODE_SYNTAX_ERROR),
				 errmsg("cannot use RETURN QUERY in a non-SETOF function")));

	if (estate->tuple_store == NULL)
		exec_init_tuple_store(estate);

	exec_run_select(estate, stmt->query, 0, &portal);

	if (!compatible_tupdesc(estate->rettupdesc, portal->tupDesc))
		ereport(ERROR,
				(errcode(ERRCODE_DATATYPE_MISMATCH),
		  errmsg("structure of query does not match function result type")));

	while (true)
	{
		MemoryContext old_cxt;
		int			i;

		SPI_cursor_fetch(portal, true, 50);
		if (SPI_processed == 0)
			break;

		old_cxt = MemoryContextSwitchTo(estate->tuple_store_cxt);
		for (i = 0; i < SPI_processed; i++)
		{
			HeapTuple	tuple = SPI_tuptable->vals[i];

			tuplestore_puttuple(estate->tuple_store, tuple);
		}
		MemoryContextSwitchTo(old_cxt);

		SPI_freetuptable(SPI_tuptable);
	}

	SPI_freetuptable(SPI_tuptable);
	SPI_cursor_close(portal);

	return PLPGSQL_RC_OK;
}

static void
exec_init_tuple_store(PLpgSQL_execstate *estate)
{
	ReturnSetInfo *rsi = estate->rsi;
	MemoryContext oldcxt;

	/*
	 * Check caller can handle a set result in the way we want
	 */
	if (!rsi || !IsA(rsi, ReturnSetInfo) ||
		(rsi->allowedModes & SFRM_Materialize) == 0 ||
		rsi->expectedDesc == NULL)
		ereport(ERROR,
				(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
				 errmsg("set-valued function called in context that cannot accept a set")));

	estate->tuple_store_cxt = rsi->econtext->ecxt_per_query_memory;

	oldcxt = MemoryContextSwitchTo(estate->tuple_store_cxt);
	estate->tuple_store = tuplestore_begin_heap(true, false, work_mem);
	MemoryContextSwitchTo(oldcxt);

	estate->rettupdesc = rsi->expectedDesc;
}

/* ----------
 * exec_stmt_raise			Build a message and throw it with elog()
 * ----------
 */
static int
exec_stmt_raise(PLpgSQL_execstate *estate, PLpgSQL_stmt_raise *stmt)
{
	char	   *cp;
	PLpgSQL_dstring ds;
	ListCell   *current_param;

	plpgsql_dstring_init(&ds);
	current_param = list_head(stmt->params);

	for (cp = stmt->message; *cp; cp++)
	{
		/*
		 * Occurrences of a single % are replaced by the next parameter's
		 * external representation. Double %'s are converted to one %.
		 */
		if (cp[0] == '%')
		{
			Oid			paramtypeid;
			Datum		paramvalue;
			bool		paramisnull;
			char	   *extval;

			if (cp[1] == '%')
			{
				plpgsql_dstring_append_char(&ds, cp[1]);
				cp++;
				continue;
			}

			if (current_param == NULL)
				ereport(ERROR,
						(errcode(ERRCODE_SYNTAX_ERROR),
						 errmsg("too few parameters specified for RAISE")));

			paramvalue = exec_eval_expr(estate,
									  (PLpgSQL_expr *) lfirst(current_param),
										&paramisnull,
										&paramtypeid);

			if (paramisnull)
				extval = "<NULL>";
			else
				extval = convert_value_to_string(paramvalue, paramtypeid);
			plpgsql_dstring_append(&ds, extval);
			current_param = lnext(current_param);
			exec_eval_cleanup(estate);
			continue;
		}

		plpgsql_dstring_append_char(&ds, cp[0]);
	}

	/*
	 * If more parameters were specified than were required to process the
	 * format string, throw an error
	 */
	if (current_param != NULL)
		ereport(ERROR,
				(errcode(ERRCODE_SYNTAX_ERROR),
				 errmsg("too many parameters specified for RAISE")));

	/*
	 * Throw the error (may or may not come back)
	 */
	estate->err_text = raise_skip_msg;	/* suppress traceback of raise */

	ereport(stmt->elog_level,
		 ((stmt->elog_level >= ERROR) ? errcode(ERRCODE_RAISE_EXCEPTION) : 0,
		  errmsg_internal("%s", plpgsql_dstring_get(&ds))));

	estate->err_text = NULL;	/* un-suppress... */

	plpgsql_dstring_free(&ds);

	return PLPGSQL_RC_OK;
}


/* ----------
 * Initialize a mostly empty execution state
 * ----------
 */
static void
plpgsql_estate_setup(PLpgSQL_execstate *estate,
					 PLpgSQL_function *func,
					 ReturnSetInfo *rsi)
{
	estate->retval = (Datum) 0;
	estate->retisnull = true;
	estate->rettype = InvalidOid;

	estate->fn_rettype = func->fn_rettype;
	estate->retistuple = func->fn_retistuple;
	estate->retisset = func->fn_retset;

	estate->readonly_func = func->fn_readonly;

	estate->rettupdesc = NULL;
	estate->exitlabel = NULL;

	estate->tuple_store = NULL;
	estate->tuple_store_cxt = NULL;
	estate->rsi = rsi;

	estate->trig_nargs = 0;
	estate->trig_argv = NULL;

	estate->found_varno = func->found_varno;
	estate->ndatums = func->ndatums;
	estate->datums = palloc(sizeof(PLpgSQL_datum *) * estate->ndatums);
	/* caller is expected to fill the datums array */

	estate->eval_tuptable = NULL;
	estate->eval_processed = 0;
	estate->eval_lastoid = InvalidOid;

	estate->err_func = func;
	estate->err_stmt = NULL;
	estate->err_text = NULL;

	/*
	 * Create an EState and ExprContext for evaluation of simple expressions.
	 */
	plpgsql_create_econtext(estate);

	/*
	 * Let the plugin see this function before we initialize any local
	 * PL/pgSQL variables - note that we also give the plugin a few function
	 * pointers so it can call back into PL/pgSQL for doing things like
	 * variable assignments and stack traces
	 */
	if (*plugin_ptr)
	{
		(*plugin_ptr)->error_callback = plpgsql_exec_error_callback;
		(*plugin_ptr)->assign_expr = exec_assign_expr;

		if ((*plugin_ptr)->func_setup)
			((*plugin_ptr)->func_setup) (estate, func);
	}
}

/* ----------
 * Release temporary memory used by expression/subselect evaluation
 *
 * NB: the result of the evaluation is no longer valid after this is done,
 * unless it is a pass-by-value datatype.
 * ----------
 */
static void
exec_eval_cleanup(PLpgSQL_execstate *estate)
{
	/* Clear result of a full SPI_execute */
	if (estate->eval_tuptable != NULL)
		SPI_freetuptable(estate->eval_tuptable);
	estate->eval_tuptable = NULL;

	/* Clear result of exec_eval_simple_expr (but keep the econtext) */
	if (estate->eval_econtext != NULL)
		ResetExprContext(estate->eval_econtext);
}


/* ----------
 * Generate a prepared plan
 * ----------
 */
static void
exec_prepare_plan(PLpgSQL_execstate *estate,
				  PLpgSQL_expr *expr, int cursorOptions)
{
	int			i;
	SPIPlanPtr	plan;
	Oid		   *argtypes;

	/*
	 * We need a temporary argtypes array to load with data. (The finished
	 * plan structure will contain a copy of it.)
	 */
	argtypes = (Oid *) palloc(expr->nparams * sizeof(Oid));

	for (i = 0; i < expr->nparams; i++)
	{
		Datum		paramval;
		bool		paramisnull;

		exec_eval_datum(estate, estate->datums[expr->params[i]],
						InvalidOid,
						&argtypes[i], &paramval, &paramisnull);
	}

	/*
	 * Generate and save the plan
	 */
	plan = SPI_prepare_cursor(expr->query, expr->nparams, argtypes,
							  cursorOptions);
	if (plan == NULL)
	{
		/* Some SPI errors deserve specific error messages */
		switch (SPI_result)
		{
			case SPI_ERROR_COPY:
				ereport(ERROR,
						(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
						 errmsg("cannot COPY to/from client in PL/pgSQL")));
			case SPI_ERROR_TRANSACTION:
				ereport(ERROR,
						(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
						 errmsg("cannot begin/end transactions in PL/pgSQL"),
						 errhint("Use a BEGIN block with an EXCEPTION clause instead.")));
			default:
				elog(ERROR, "SPI_prepare_cursor failed for \"%s\": %s",
					 expr->query, SPI_result_code_string(SPI_result));
		}
	}
	expr->plan = SPI_saveplan(plan);
	SPI_freeplan(plan);
	plan = expr->plan;
	expr->plan_argtypes = plan->argtypes;
	exec_simple_check_plan(expr);

	pfree(argtypes);
}


/* ----------
 * exec_stmt_execsql			Execute an SQL statement (possibly with INTO).
 * ----------
 */
static int
exec_stmt_execsql(PLpgSQL_execstate *estate,
				  PLpgSQL_stmt_execsql *stmt)
{
	int			i;
	Datum	   *values;
	char	   *nulls;
	long		tcount;
	int			rc;
	PLpgSQL_expr *expr = stmt->sqlstmt;

	/*
	 * On the first call for this statement generate the plan, and detect
	 * whether the statement is INSERT/UPDATE/DELETE
	 */
	if (expr->plan == NULL)
	{
		ListCell   *l;

		exec_prepare_plan(estate, expr, 0);
		stmt->mod_stmt = false;
		foreach(l, expr->plan->plancache_list)
		{
			CachedPlanSource *plansource = (CachedPlanSource *) lfirst(l);
			ListCell   *l2;

			foreach(l2, plansource->plan->stmt_list)
			{
				PlannedStmt *p = (PlannedStmt *) lfirst(l2);

				if (IsA(p, PlannedStmt) &&
					p->canSetTag)
				{
					if (p->commandType == CMD_INSERT ||
						p->commandType == CMD_UPDATE ||
						p->commandType == CMD_DELETE)
						stmt->mod_stmt = true;
				}
			}
		}
	}

	/*
	 * Now build up the values and nulls arguments for SPI_execute_plan()
	 */
	values = (Datum *) palloc(expr->nparams * sizeof(Datum));
	nulls = (char *) palloc(expr->nparams * sizeof(char));

	for (i = 0; i < expr->nparams; i++)
	{
		PLpgSQL_datum *datum = estate->datums[expr->params[i]];
		Oid			paramtypeid;
		bool		paramisnull;

		exec_eval_datum(estate, datum, expr->plan_argtypes[i],
						&paramtypeid, &values[i], &paramisnull);
		if (paramisnull)
			nulls[i] = 'n';
		else
			nulls[i] = ' ';
	}

	/*
	 * If we have INTO, then we only need one row back ... but if we have INTO
	 * STRICT, ask for two rows, so that we can verify the statement returns
	 * only one.  INSERT/UPDATE/DELET