Additional SPI functions

diff --git a/src/backend/executor/spi.c b/src/backend/executor/spi.c
index 4a6b966..4bfc4f4 100644
--- a/src/backend/executor/spi.c
+++ b/src/backend/executor/spi.c
@@ -22,6 +22,8 @@
 #include "commands/trigger.h"
 #include "executor/executor.h"
 #include "executor/spi_priv.h"
+#include "nodes/pg_list.h"
+#include "parser/analyze.h"
 #include "tcop/pquery.h"
 #include "tcop/utility.h"
 #include "utils/builtins.h"
@@ -31,6 +33,7 @@
 #include "utils/snapmgr.h"
 #include "utils/syscache.h"
 #include "utils/typcache.h"
+#include "rewrite/rewriteHandler.h"
 
 
 uint32		SPI_processed = 0;
@@ -50,6 +53,11 @@ static Portal SPI_cursor_open_internal(const char *name, SPIPlanPtr plan,
 static void _SPI_prepare_plan(const char *src, SPIPlanPtr plan,
 				  ParamListInfo boundParams);
 
+static int _SPI_prepare_statement_plan(const char *src, SPIPlanPtr plan,
+				  SPIParamCallback pcb, void *cb_data);
+
+static int _SPI_execute_statements(const char *src);
+
 static int _SPI_execute_plan(SPIPlanPtr plan, ParamListInfo paramLI,
 				  Snapshot snapshot, Snapshot crosscheck_snapshot,
 				  bool read_only, bool fire_triggers, long tcount);
@@ -524,6 +532,55 @@ SPI_execute_with_args(const char *src,
 }
 
 SPIPlanPtr
+SPI_prepare_statement(const char *src, int cursorOptions, SPIParamCallback pcb, void *cbdata, TupleDesc *resultDesc)
+{
+	_SPI_plan	plan;
+	SPIPlanPtr	rplan;
+
+	SPI_result = _SPI_begin_call(true);
+	if (SPI_result < 0)
+		return NULL;
+
+	memset(&plan, 0, sizeof(_SPI_plan));
+	plan.magic = _SPI_PLAN_MAGIC;
+	plan.cursor_options = cursorOptions;
+	plan.nargs = -1;
+	plan.argtypes = NULL;
+
+	SPI_result = _SPI_prepare_statement_plan(src, &plan, pcb, cbdata);
+
+	if (SPI_result < 0)
+		rplan = NULL;
+	else
+	{
+		rplan = _SPI_copy_plan(&plan, _SPI_current->procCxt);
+		if (rplan != NULL)
+			*resultDesc =
+				((CachedPlanSource *) linitial(rplan->plancache_list))->resultDesc;
+	}
+
+	_SPI_end_call(true);
+
+	return rplan;
+}
+
+int
+SPI_execute_statements(const char *src)
+{
+	int r = -1;
+
+	SPI_result = _SPI_begin_call(true);
+	if (SPI_result < 0)
+		return r;
+
+	SPI_result = r = _SPI_execute_statements(src);
+
+	_SPI_end_call(true);
+
+	return r;
+}
+
+SPIPlanPtr
 SPI_prepare(const char *src, int nargs, Oid *argtypes)
 {
 	return SPI_prepare_cursor(src, nargs, argtypes, 0);
@@ -1648,6 +1705,278 @@ spi_printtup(TupleTableSlot *slot, DestReceiver *self)
  */
 
 /*
+ * Parse and plan a *single* statement.
+ *
+ * Given a destination plan, parse and plan the given statement string into the
+ * SPIPlanPtr.
+ *
+ * In contrast to _SPI_prepare_plan, this function will analyze the raw parse
+ * tree and extract the statement's parameters from the query. After collecting
+ * the parameter types, the SPIParamCallback is invoked to allow the original caller
+ * to note the parameter types, and to provide actual parameters if so desired.
+ *
+ * This requires that the given SQL is a single statement.
+ * Transaction and STDIN/STDOUT COPY statements are restricted.
+ */
+static int
+_SPI_prepare_statement_plan(const char *src, SPIPlanPtr plan, SPIParamCallback spipcb, void *cbdata)
+{
+	List	   *raw_parsetree_list;
+	Node	   *rpt;
+	List	   *plancache_list;
+	Query	   *query;
+	ErrorContextCallback spierrcontext;
+	Oid		   *argtypes = NULL;
+	int			nargs = -1;
+	int			cursor_options = plan->cursor_options;
+	List	   *stmt_list;
+	CachedPlanSource	*plansource;
+	CachedPlan	*cplan;
+	Datum *param_datums = NULL;
+	bool *param_nulls = NULL;
+	ParamListInfo	boundParams;
+	const char *commandTag;
+
+	/*
+	 * Setup error traceback support for ereport()
+	 */
+	spierrcontext.callback = _SPI_error_callback;
+	spierrcontext.arg = (void *) src;
+	spierrcontext.previous = error_context_stack;
+	error_context_stack = &spierrcontext;
+
+	/*
+	 * Parse the request string into a list of raw parse trees.
+	 */
+	raw_parsetree_list = pg_parse_query(src);
+	if (list_length(raw_parsetree_list) != 1)
+	{
+		ereport(ERROR,
+				(errcode(ERRCODE_SYNTAX_ERROR),
+				errmsg("cannot insert multiple commands into a prepared statement")));
+	}
+
+	rpt = (Node *) linitial(raw_parsetree_list);
+
+	/*
+	 * Validate that it fits the basic criteria for a single SPI statement.
+	 *
+	 * Don't wait for execution time to fail on these.
+	 */
+	if (IsA(rpt, TransactionStmt))
+	{
+		SPI_result = SPI_ERROR_TRANSACTION;
+		goto fail;
+	}
+	if (IsA(rpt, CopyStmt) && ((CopyStmt *) rpt)->filename == NULL)
+	{
+		SPI_result = SPI_ERROR_COPY;
+		goto fail;
+	}
+
+	/*
+	 * for repalloc
+	 */
+	nargs = 0;
+	argtypes = palloc(sizeof(Oid) * 0);
+
+	/* Need a copyObject here to keep parser from modifying raw tree */
+	query = parse_analyze_varparams(
+						(Node *) copyObject(rpt), src, &argtypes, &nargs);
+
+	plan->argtypes = argtypes;
+	plan->nargs = nargs;
+
+	commandTag = CreateCommandTag(rpt);
+	/*
+	 * Run the callback to give the caller the parameter
+	 * type Oids and, potentially, collect parameters for the statement.
+	 */
+	spipcb(cbdata, commandTag, nargs, argtypes, &param_datums, &param_nulls);
+
+	/*
+	 * If the callback set some parameters, use them as constants in the plan.
+	 */
+	if (PointerIsValid(param_datums))
+	{
+		if (!PointerIsValid(param_nulls))
+		{
+			SPI_result = SPI_ERROR_ARGUMENT;
+			goto fail;
+		}
+
+		boundParams = _SPI_convert_params(nargs, argtypes,
+									  param_datums, param_nulls,
+									  PARAM_FLAG_CONST);
+	}
+	else
+		boundParams = NULL;
+
+	/*
+	 * Don't rewrite utility statements.
+	 */
+	if (query->commandType == CMD_UTILITY)
+		stmt_list = list_make1(query);
+	else
+		stmt_list = QueryRewrite(query);
+
+	stmt_list = pg_plan_queries(stmt_list, cursor_options, boundParams);
+
+	plansource = (CachedPlanSource *) palloc0(sizeof(CachedPlanSource));
+	cplan = (CachedPlan *) palloc0(sizeof(CachedPlan));
+
+	plansource->raw_parse_tree = rpt;
+	plansource->query_string = pstrdup(src);
+	plansource->commandTag = commandTag;
+	plansource->param_types = argtypes;
+	plansource->num_params = nargs;
+	plansource->fully_planned = true;
+	plansource->fixed_result = true;
+	plansource->resultDesc = PlanCacheComputeResultDesc(stmt_list);
+	plansource->plan = cplan;
+
+	cplan->stmt_list = stmt_list;
+	cplan->fully_planned = true;
+
+	plancache_list = NIL;
+	plancache_list = lappend(plancache_list, plansource);
+	plan->plancache_list = plancache_list;
+
+	/*
+	 * Pop the error context stack
+	 */
+	error_context_stack = spierrcontext.previous;
+
+	return 0;
+fail:
+	error_context_stack = spierrcontext.previous;
+	return SPI_result;
+}
+
+/*
+ * Parse, plan and execute all the statements in the given string.
+ * Execution occurs directly after a statement is planned and *before* the next
+ * statement is planned. This allows statements to depend on objects created by
+ * statements executed prior.
+ *
+ * This interface is designed for allowing users to execute bulk DDL/DML.
+ */
+static int
+_SPI_execute_statements(const char *src)
+{
+	int		r = 0;
+	List	   *raw_parsetree_list;
+	ListCell   *list_item;
+	DestReceiver *dest;
+	ErrorContextCallback spierrcontext;
+
+	/*
+	 * Setup error traceback support for ereport()
+	 */
+	spierrcontext.callback = _SPI_error_callback;
+	spierrcontext.arg = (void *) src;
+	spierrcontext.previous = error_context_stack;
+	error_context_stack = &spierrcontext;
+
+	/*
+	 * Parse the request string into a list of raw parse trees.
+	 */
+	raw_parsetree_list = pg_parse_query(src);
+
+	/*
+	 * Everything is sent to DestNone.
+	 */
+	_SPI_current->processed = 0;
+	_SPI_current->lastoid = InvalidOid;
+	_SPI_current->tuptable = NULL;
+	dest = CreateDestReceiver(DestNone);
+
+	/*
+	 * Iterate over the parsetree list, executing each statement.
+	 */
+	foreach(list_item, raw_parsetree_list)
+	{
+		Node	   *parsetree = (Node *) lfirst(list_item);
+		List	   *stmt_list;
+		ListCell	*lc2;
+
+		/*
+		 * Before each statement.
+		 */
+		CommandCounterIncrement();
+
+		/*
+		 * No parameters.
+		 */
+		stmt_list = pg_analyze_and_rewrite(parsetree, src, NULL, 0);
+		stmt_list = pg_plan_queries(stmt_list, 0, NULL);
+
+		foreach(lc2, stmt_list)
+		{
+			Node	   *stmt = (Node *) lfirst(lc2);
+
+			if (!IsA(stmt, PlannedStmt))
+			{
+				/*
+				 * Filter prohibited statements.
+				 */
+				if (IsA(stmt, CopyStmt))
+				{
+					CopyStmt   *cstmt = (CopyStmt *) stmt;
+
+					if (cstmt->filename == NULL)
+					{
+						return SPI_ERROR_COPY;
+					}
+				}
+				else if (IsA(stmt, TransactionStmt))
+				{
+					return SPI_ERROR_TRANSACTION;
+				}
+			}
+
+			/*
+			 * Before each command.
+			 */
+			CommandCounterIncrement();
+			PushActiveSnapshot(GetTransactionSnapshot());
+
+			if (IsA(stmt, PlannedStmt) &&
+				((PlannedStmt *) stmt)->utilityStmt == NULL)
+			{
+				QueryDesc  *qdesc;
+
+				qdesc = CreateQueryDesc(
+										(PlannedStmt *) stmt, src,
+										GetActiveSnapshot(), /* xact snapshot */
+										InvalidSnapshot, /* no crosscheck */
+										dest, NULL, false);
+				r = _SPI_pquery(qdesc, true, 0);
+				FreeQueryDesc(qdesc);
+			}
+			else
+			{
+				ProcessUtility(stmt, src,
+							   NULL,	/* no params */
+							   false,	/* not top level */
+							   dest,		/* DestNone */
+							   NULL);
+				r = SPI_OK_UTILITY;
+			}
+
+			PopActiveSnapshot();
+
+			if (r < 0)
+				return r;
+		}
+	}
+
+	error_context_stack = spierrcontext.previous;
+
+	return r;
+}
+
+/*
  * Parse and plan a querystring.
  *
  * At entry, plan->argtypes and plan->nargs (or alternatively plan->parserSetup
diff --git a/src/backend/utils/error/elog.c b/src/backend/utils/error/elog.c
index 025ca35..5b7ffa2 100644
--- a/src/backend/utils/error/elog.c
+++ b/src/backend/utils/error/elog.c
@@ -1021,6 +1021,23 @@ internalerrquery(const char *query)
 }
 
 /*
+ * getelevel --- return the currently set elevel
+ *
+ * This is only intended for use in error callback subroutines, since there
+ * is no other place outside elog.c where the concept is meaningful.
+ */
+int
+getelevel(void)
+{
+	ErrorData  *edata = &errordata[errordata_stack_depth];
+
+	/* we don't bother incrementing recursion_depth */
+	CHECK_STACK_DEPTH();
+
+	return edata->elevel;
+}
+
+/*
  * geterrcode --- return the currently set SQLSTATE error code
  *
  * This is only intended for use in error callback subroutines, since there
diff --git a/src/include/executor/spi.h b/src/include/executor/spi.h
index 33f4f15..b23f7a1 100644
--- a/src/include/executor/spi.h
+++ b/src/include/executor/spi.h
@@ -31,6 +31,16 @@ typedef struct SPITupleTable
 /* Plans are opaque structs for standard users of SPI */
 typedef struct _SPI_plan *SPIPlanPtr;
 
+/*
+ * Callback function used by SPI_prepare_statement to give the
+ * caller the type Oid's of the statement's parameters.
+ *
+ * Additionally, the caller has the ability to give the param_values.
+ */
+typedef void (*SPIParamCallback)(
+	void *cb_data, const char *commandTag,
+	int nargs, Oid *typoids, Datum **param_values, char **param_nulls);
+
 #define SPI_ERROR_CONNECT		(-1)
 #define SPI_ERROR_COPY			(-2)
 #define SPI_ERROR_OPUNKNOWN		(-3)
@@ -88,6 +98,9 @@ extern int SPI_execute_with_args(const char *src,
 					  int nargs, Oid *argtypes,
 					  Datum *Values, const char *Nulls,
 					  bool read_only, long tcount);
+extern int SPI_execute_statements(const char *src);
+extern SPIPlanPtr SPI_prepare_statement(const char *src, int cursorOptions,
+					 SPIParamCallback pcb, void *pcb_arg, TupleDesc *resultDesc);
 extern SPIPlanPtr SPI_prepare(const char *src, int nargs, Oid *argtypes);
 extern SPIPlanPtr SPI_prepare_cursor(const char *src, int nargs, Oid *argtypes,
 				   int cursorOptions);
diff --git a/src/include/utils/elog.h b/src/include/utils/elog.h
index 0982ca4..ea5a1c1 100644
--- a/src/include/utils/elog.h
+++ b/src/include/utils/elog.h
@@ -180,6 +180,7 @@ extern int	errposition(int cursorpos);
 extern int	internalerrposition(int cursorpos);
 extern int	internalerrquery(const char *query);
 
+extern int	getelevel(void);
 extern int	geterrcode(void);
 extern int	geterrposition(void);
 extern int	getinternalerrposition(void);