PL/pgSQL nested CALL with transactions

From 2254a08df89db4191ed63753f95bb8fbda5ef150 Mon Sep 17 00:00:00 2001
From: Peter Eisentraut <peter_e@gmx.net>
Date: Thu, 15 Mar 2018 21:54:28 -0400
Subject: [PATCH v2] PL/pgSQL: Nested CALL with transactions

So far, a nested CALL or DO in PL/pgSQL would not establish a context
where transaction control statements were allowed.  This fixes that by
handling CALL and DO specially in PL/pgSQL, passing the atomic/nonatomic
execution context through and doing the required management around
transaction boundaries.

XXX This requires a new flag in SPI to run SPI_execute* without snapshot
management.  This currently poked directly into the plan struct,
requiring access to spi_priv.h.  This could use some refinement ideas.
---
 doc/src/sgml/plpgsql.sgml                          | 13 +++-
 src/backend/executor/spi.c                         | 12 ++--
 src/backend/tcop/utility.c                         |  4 +-
 src/include/executor/spi_priv.h                    |  1 +
 src/include/tcop/utility.h                         |  1 +
 .../plpgsql/src/expected/plpgsql_transaction.out   | 71 +++++++++++++++++-----
 src/pl/plpgsql/src/pl_exec.c                       | 56 +++++++++++++++--
 src/pl/plpgsql/src/pl_funcs.c                      |  4 +-
 src/pl/plpgsql/src/pl_gram.y                       | 18 ++++++
 src/pl/plpgsql/src/pl_handler.c                    |  4 +-
 src/pl/plpgsql/src/pl_scanner.c                    |  1 +
 src/pl/plpgsql/src/plpgsql.h                       |  5 +-
 src/pl/plpgsql/src/sql/plpgsql_transaction.sql     | 42 +++++++++++--
 13 files changed, 188 insertions(+), 44 deletions(-)

diff --git a/doc/src/sgml/plpgsql.sgml b/doc/src/sgml/plpgsql.sgml
index 6c25116538..ba3117c733 100644
--- a/doc/src/sgml/plpgsql.sgml
+++ b/doc/src/sgml/plpgsql.sgml
@@ -3463,9 +3463,9 @@ <title>Looping Through a Cursor's Result</title>
    <title>Transaction Management</title>
 
    <para>
-    In procedures invoked by the <command>CALL</command> command from the top
-    level as well as in anonymous code blocks (<command>DO</command> command)
-    called from the top level, it is possible to end transactions using the
+    In procedures invoked by the <command>CALL</command> command
+    as well as in anonymous code blocks (<command>DO</command> command),
+    it is possible to end transactions using the
     commands <command>COMMIT</command> and <command>ROLLBACK</command>.  A new
     transaction is started automatically after a transaction is ended using
     these commands, so there is no separate <command>START
@@ -3495,6 +3495,13 @@ <title>Transaction Management</title>
 </programlisting>
    </para>
 
+   <para>
+    Transaction control is only possible in <command>CALL</command> or
+    <command>DO</command> invocations from the top level or nested
+    <command>CALL</command> or <command>DO</command> invocations without any
+    other intervening command.
+   </para>
+
    <para>
     A transaction cannot be ended inside a loop over a query result, nor
     inside a block with exception handlers.
diff --git a/src/backend/executor/spi.c b/src/backend/executor/spi.c
index 9fc4431b80..174ec6cd5a 100644
--- a/src/backend/executor/spi.c
+++ b/src/backend/executor/spi.c
@@ -2042,7 +2042,7 @@ _SPI_execute_plan(SPIPlanPtr plan, ParamListInfo paramLI,
 	 * In the first two cases, we can just push the snap onto the stack once
 	 * for the whole plan list.
 	 */
-	if (snapshot != InvalidSnapshot)
+	if (snapshot != InvalidSnapshot && !plan->no_snapshots)
 	{
 		if (read_only)
 		{
@@ -2121,7 +2121,7 @@ _SPI_execute_plan(SPIPlanPtr plan, ParamListInfo paramLI,
 		 * In the default non-read-only case, get a new snapshot, replacing
 		 * any that we pushed in a previous cycle.
 		 */
-		if (snapshot == InvalidSnapshot && !read_only)
+		if (snapshot == InvalidSnapshot && !read_only && !plan->no_snapshots)
 		{
 			if (pushed_active_snap)
 				PopActiveSnapshot();
@@ -2172,7 +2172,7 @@ _SPI_execute_plan(SPIPlanPtr plan, ParamListInfo paramLI,
 			 * If not read-only mode, advance the command counter before each
 			 * command and update the snapshot.
 			 */
-			if (!read_only)
+			if (!read_only && !plan->no_snapshots)
 			{
 				CommandCounterIncrement();
 				UpdateActiveSnapshotCommandId();
@@ -2206,7 +2206,7 @@ _SPI_execute_plan(SPIPlanPtr plan, ParamListInfo paramLI,
 
 				ProcessUtility(stmt,
 							   plansource->query_string,
-							   PROCESS_UTILITY_QUERY,
+							   _SPI_current->atomic ? PROCESS_UTILITY_QUERY : PROCESS_UTILITY_QUERY_NONATOMIC,
 							   paramLI,
 							   _SPI_current->queryEnv,
 							   dest,
@@ -2638,7 +2638,7 @@ _SPI_make_plan_non_temp(SPIPlanPtr plan)
 	oldcxt = MemoryContextSwitchTo(plancxt);
 
 	/* Copy the SPI_plan struct and subsidiary data into the new context */
-	newplan = (SPIPlanPtr) palloc(sizeof(_SPI_plan));
+	newplan = (SPIPlanPtr) palloc0(sizeof(_SPI_plan));
 	newplan->magic = _SPI_PLAN_MAGIC;
 	newplan->saved = false;
 	newplan->oneshot = false;
@@ -2705,7 +2705,7 @@ _SPI_save_plan(SPIPlanPtr plan)
 	oldcxt = MemoryContextSwitchTo(plancxt);
 
 	/* Copy the SPI plan into its own context */
-	newplan = (SPIPlanPtr) palloc(sizeof(_SPI_plan));
+	newplan = (SPIPlanPtr) palloc0(sizeof(_SPI_plan));
 	newplan->magic = _SPI_PLAN_MAGIC;
 	newplan->saved = false;
 	newplan->oneshot = false;
diff --git a/src/backend/tcop/utility.c b/src/backend/tcop/utility.c
index 6effe031f8..e9823301d8 100644
--- a/src/backend/tcop/utility.c
+++ b/src/backend/tcop/utility.c
@@ -531,7 +531,7 @@ standard_ProcessUtility(PlannedStmt *pstmt,
 
 		case T_DoStmt:
 			ExecuteDoStmt((DoStmt *) parsetree,
-						  (context != PROCESS_UTILITY_TOPLEVEL || IsTransactionBlock()));
+						  (!(context == PROCESS_UTILITY_TOPLEVEL || context == PROCESS_UTILITY_QUERY_NONATOMIC) || IsTransactionBlock()));
 			break;
 
 		case T_CreateTableSpaceStmt:
@@ -661,7 +661,7 @@ standard_ProcessUtility(PlannedStmt *pstmt,
 
 		case T_CallStmt:
 			ExecuteCallStmt(castNode(CallStmt, parsetree), params,
-							(context != PROCESS_UTILITY_TOPLEVEL || IsTransactionBlock()),
+							(!(context == PROCESS_UTILITY_TOPLEVEL || context == PROCESS_UTILITY_QUERY_NONATOMIC) || IsTransactionBlock()),
 							dest);
 			break;
 
diff --git a/src/include/executor/spi_priv.h b/src/include/executor/spi_priv.h
index 263c8f1453..376fae0bbc 100644
--- a/src/include/executor/spi_priv.h
+++ b/src/include/executor/spi_priv.h
@@ -86,6 +86,7 @@ typedef struct _SPI_plan
 	int			magic;			/* should equal _SPI_PLAN_MAGIC */
 	bool		saved;			/* saved or unsaved plan? */
 	bool		oneshot;		/* one-shot plan? */
+	bool		no_snapshots;	/* let the caller handle the snapshots */
 	List	   *plancache_list; /* one CachedPlanSource per parsetree */
 	MemoryContext plancxt;		/* Context containing _SPI_plan and data */
 	int			cursor_options; /* Cursor options used for planning */
diff --git a/src/include/tcop/utility.h b/src/include/tcop/utility.h
index 5550055710..880d19311a 100644
--- a/src/include/tcop/utility.h
+++ b/src/include/tcop/utility.h
@@ -20,6 +20,7 @@ typedef enum
 {
 	PROCESS_UTILITY_TOPLEVEL,	/* toplevel interactive command */
 	PROCESS_UTILITY_QUERY,		/* a complete query, but not toplevel */
+	PROCESS_UTILITY_QUERY_NONATOMIC, /* a complete query, nonatomic execution context */
 	PROCESS_UTILITY_SUBCOMMAND	/* a portion of a query */
 } ProcessUtilityContext;
 
diff --git a/src/pl/plpgsql/src/expected/plpgsql_transaction.out b/src/pl/plpgsql/src/expected/plpgsql_transaction.out
index ce66487137..583eac46a9 100644
--- a/src/pl/plpgsql/src/expected/plpgsql_transaction.out
+++ b/src/pl/plpgsql/src/expected/plpgsql_transaction.out
@@ -1,10 +1,10 @@
 CREATE TABLE test1 (a int, b text);
-CREATE PROCEDURE transaction_test1()
+CREATE PROCEDURE transaction_test1(x int, y text)
 LANGUAGE plpgsql
 AS $$
 BEGIN
-    FOR i IN 0..9 LOOP
-        INSERT INTO test1 (a) VALUES (i);
+    FOR i IN 0..x LOOP
+        INSERT INTO test1 (a, b) VALUES (i, y);
         IF i % 2 = 0 THEN
             COMMIT;
         ELSE
@@ -13,15 +13,15 @@ BEGIN
     END LOOP;
 END
 $$;
-CALL transaction_test1();
+CALL transaction_test1(9, 'foo');
 SELECT * FROM test1;
- a | b 
----+---
- 0 | 
- 2 | 
- 4 | 
- 6 | 
- 8 | 
+ a |  b  
+---+-----
+ 0 | foo
+ 2 | foo
+ 4 | foo
+ 6 | foo
+ 8 | foo
 (5 rows)
 
 TRUNCATE test1;
@@ -51,9 +51,9 @@ SELECT * FROM test1;
 
 -- transaction commands not allowed when called in transaction block
 START TRANSACTION;
-CALL transaction_test1();
+CALL transaction_test1(9, 'error');
 ERROR:  invalid transaction termination
-CONTEXT:  PL/pgSQL function transaction_test1() line 6 at COMMIT
+CONTEXT:  PL/pgSQL function transaction_test1(integer,text) line 6 at COMMIT
 COMMIT;
 START TRANSACTION;
 DO LANGUAGE plpgsql $$ BEGIN COMMIT; END $$;
@@ -90,14 +90,14 @@ CREATE FUNCTION transaction_test3() RETURNS int
 LANGUAGE plpgsql
 AS $$
 BEGIN
-    CALL transaction_test1();
+    CALL transaction_test1(9, 'error');
     RETURN 1;
 END;
 $$;
 SELECT transaction_test3();
 ERROR:  invalid transaction termination
-CONTEXT:  PL/pgSQL function transaction_test1() line 6 at COMMIT
-SQL statement "CALL transaction_test1()"
+CONTEXT:  PL/pgSQL function transaction_test1(integer,text) line 6 at COMMIT
+SQL statement "CALL transaction_test1(9, 'error')"
 PL/pgSQL function transaction_test3() line 3 at CALL
 SELECT * FROM test1;
  a | b 
@@ -130,6 +130,45 @@ $$;
 CALL transaction_test5();
 ERROR:  invalid transaction termination
 CONTEXT:  PL/pgSQL function transaction_test5() line 3 at COMMIT
+TRUNCATE test1;
+-- nested procedure calls
+CREATE PROCEDURE transaction_test6(c text)
+LANGUAGE plpgsql
+AS $$
+BEGIN
+    CALL transaction_test1(9, c);
+END;
+$$;
+CALL transaction_test6('bar');
+SELECT * FROM test1;
+ a |  b  
+---+-----
+ 0 | bar
+ 2 | bar
+ 4 | bar
+ 6 | bar
+ 8 | bar
+(5 rows)
+
+TRUNCATE test1;
+CREATE PROCEDURE transaction_test7()
+LANGUAGE plpgsql
+AS $$
+BEGIN
+    DO 'BEGIN CALL transaction_test1(9, $x$baz$x$); END;';
+END;
+$$;
+CALL transaction_test7();
+SELECT * FROM test1;
+ a |  b  
+---+-----
+ 0 | baz
+ 2 | baz
+ 4 | baz
+ 6 | baz
+ 8 | baz
+(5 rows)
+
 -- commit inside cursor loop
 CREATE TABLE test2 (x int);
 INSERT INTO test2 VALUES (0), (1), (2), (3), (4);
diff --git a/src/pl/plpgsql/src/pl_exec.c b/src/pl/plpgsql/src/pl_exec.c
index 68da7cf669..8a747880e4 100644
--- a/src/pl/plpgsql/src/pl_exec.c
+++ b/src/pl/plpgsql/src/pl_exec.c
@@ -22,6 +22,7 @@
 #include "access/tupconvert.h"
 #include "catalog/pg_proc.h"
 #include "catalog/pg_type.h"
+#include "commands/defrem.h"
 #include "executor/execExpr.h"
 #include "executor/spi.h"
 #include "executor/spi_priv.h"
@@ -33,6 +34,7 @@
 #include "parser/scansup.h"
 #include "storage/proc.h"
 #include "tcop/tcopprot.h"
+#include "tcop/utility.h"
 #include "utils/array.h"
 #include "utils/builtins.h"
 #include "utils/datum.h"
@@ -440,7 +442,7 @@ static char *format_preparedparamsdata(PLpgSQL_execstate *estate,
  */
 Datum
 plpgsql_exec_function(PLpgSQL_function *func, FunctionCallInfo fcinfo,
-					  EState *simple_eval_estate)
+					  EState *simple_eval_estate, bool atomic)
 {
 	PLpgSQL_execstate estate;
 	ErrorContextCallback plerrcontext;
@@ -452,6 +454,7 @@ plpgsql_exec_function(PLpgSQL_function *func, FunctionCallInfo fcinfo,
 	 */
 	plpgsql_estate_setup(&estate, func, (ReturnSetInfo *) fcinfo->resultinfo,
 						 simple_eval_estate);
+	estate.atomic = atomic;
 
 	/*
 	 * Setup error traceback support for ereport()
@@ -2057,20 +2060,58 @@ exec_stmt_call(PLpgSQL_execstate *estate, PLpgSQL_stmt_call *stmt)
 {
 	PLpgSQL_expr *expr = stmt->expr;
 	ParamListInfo paramLI;
+	LocalTransactionId before_lxid;
+	LocalTransactionId after_lxid;
 	int			rc;
 
 	if (expr->plan == NULL)
-		exec_prepare_plan(estate, expr, 0);
+	{
+		SPIPlanPtr	plan;
+
+		expr->func = estate->func;
+
+		plan = SPI_prepare_params(expr->query,
+								  (ParserSetupHook) plpgsql_parser_setup,
+								  (void *) expr,
+								  0);
+
+		if (plan == NULL)
+			elog(ERROR, "SPI_prepare_params failed for \"%s\": %s",
+				 expr->query, SPI_result_code_string(SPI_result));
+
+		/*
+		 * The procedure call could end transactions, which would upset the
+		 * snapshot management in SPI_execute*, so don't let it do it.
+		 */
+		plan->no_snapshots = true;
+
+		expr->plan = plan;
+		expr->rwparam = -1;
+	}
 
 	paramLI = setup_param_list(estate, expr);
 
+	before_lxid = MyProc->lxid;
+
 	rc = SPI_execute_plan_with_paramlist(expr->plan, paramLI,
 										 estate->readonly_func, 0);
 
+	after_lxid = MyProc->lxid;
+
 	if (rc < 0)
 		elog(ERROR, "SPI_execute_plan_with_paramlist failed executing query \"%s\": %s",
 			 expr->query, SPI_result_code_string(rc));
 
+	/*
+	 * If we are in a new transaction after the call, we need to reset some
+	 * internal state.
+	 */
+	if (before_lxid != after_lxid)
+	{
+		estate->simple_eval_estate = NULL;
+		plpgsql_create_econtext(estate);
+	}
+
 	if (SPI_processed == 1)
 	{
 		SPITupleTable *tuptab = SPI_tuptable;
@@ -3719,6 +3760,7 @@ plpgsql_estate_setup(PLpgSQL_execstate *estate,
 	estate->retisset = func->fn_retset;
 
 	estate->readonly_func = func->fn_readonly;
+	estate->atomic = true;
 
 	estate->exitlabel = NULL;
 	estate->cur_error = NULL;
@@ -7834,11 +7876,13 @@ plpgsql_create_econtext(PLpgSQL_execstate *estate)
 	{
 		MemoryContext oldcontext;
 
-		Assert(shared_simple_eval_estate == NULL);
-		oldcontext = MemoryContextSwitchTo(TopTransactionContext);
-		shared_simple_eval_estate = CreateExecutorState();
+		if (shared_simple_eval_estate == NULL)
+		{
+			oldcontext = MemoryContextSwitchTo(TopTransactionContext);
+			shared_simple_eval_estate = CreateExecutorState();
+			MemoryContextSwitchTo(oldcontext);
+		}
 		estate->simple_eval_estate = shared_simple_eval_estate;
-		MemoryContextSwitchTo(oldcontext);
 	}
 
 	/*
diff --git a/src/pl/plpgsql/src/pl_funcs.c b/src/pl/plpgsql/src/pl_funcs.c
index 39d6a54663..fc96fb5f4d 100644
--- a/src/pl/plpgsql/src/pl_funcs.c
+++ b/src/pl/plpgsql/src/pl_funcs.c
@@ -285,7 +285,7 @@ plpgsql_stmt_typename(PLpgSQL_stmt *stmt)
 		case PLPGSQL_STMT_PERFORM:
 			return "PERFORM";
 		case PLPGSQL_STMT_CALL:
-			return "CALL";
+			return ((PLpgSQL_stmt_call *) stmt)->is_call ? "CALL" : "DO";
 		case PLPGSQL_STMT_COMMIT:
 			return "COMMIT";
 		case PLPGSQL_STMT_ROLLBACK:
@@ -1295,7 +1295,7 @@ static void
 dump_call(PLpgSQL_stmt_call *stmt)
 {
 	dump_ind();
-	printf("CALL expr = ");
+	printf("%s expr = ", stmt->is_call ? "CALL" : "DO");
 	dump_expr(stmt->expr);
 	printf("\n");
 }
diff --git a/src/pl/plpgsql/src/pl_gram.y b/src/pl/plpgsql/src/pl_gram.y
index 4c80936678..b8562ca8b4 100644
--- a/src/pl/plpgsql/src/pl_gram.y
+++ b/src/pl/plpgsql/src/pl_gram.y
@@ -276,6 +276,7 @@ static	void			check_raise_parameters(PLpgSQL_stmt_raise *stmt);
 %token <keyword>	K_DEFAULT
 %token <keyword>	K_DETAIL
 %token <keyword>	K_DIAGNOSTICS
+%token <keyword>	K_DO
 %token <keyword>	K_DUMP
 %token <keyword>	K_ELSE
 %token <keyword>	K_ELSIF
@@ -914,8 +915,24 @@ stmt_call		: K_CALL
 						new->cmd_type = PLPGSQL_STMT_CALL;
 						new->lineno = plpgsql_location_to_lineno(@1);
 						new->expr = read_sql_stmt("CALL ");
+						new->is_call = true;
 
 						$$ = (PLpgSQL_stmt *)new;
+
+					}
+				| K_DO
+					{
+						/* use the same structures as for CALL, for simplicity */
+						PLpgSQL_stmt_call *new;
+
+						new = palloc0(sizeof(PLpgSQL_stmt_call));
+						new->cmd_type = PLPGSQL_STMT_CALL;
+						new->lineno = plpgsql_location_to_lineno(@1);
+						new->expr = read_sql_stmt("DO ");
+						new->is_call = false;
+
+						$$ = (PLpgSQL_stmt *)new;
+
 					}
 				;
 
@@ -2434,6 +2451,7 @@ unreserved_keyword	:
 				| K_DEFAULT
 				| K_DETAIL
 				| K_DIAGNOSTICS
+				| K_DO
 				| K_DUMP
 				| K_ELSIF
 				| K_ERRCODE
diff --git a/src/pl/plpgsql/src/pl_handler.c b/src/pl/plpgsql/src/pl_handler.c
index f38ef04077..61452d9f7f 100644
--- a/src/pl/plpgsql/src/pl_handler.c
+++ b/src/pl/plpgsql/src/pl_handler.c
@@ -260,7 +260,7 @@ plpgsql_call_handler(PG_FUNCTION_ARGS)
 			retval = (Datum) 0;
 		}
 		else
-			retval = plpgsql_exec_function(func, fcinfo, NULL);
+			retval = plpgsql_exec_function(func, fcinfo, NULL, !nonatomic);
 	}
 	PG_CATCH();
 	{
@@ -332,7 +332,7 @@ plpgsql_inline_handler(PG_FUNCTION_ARGS)
 	/* And run the function */
 	PG_TRY();
 	{
-		retval = plpgsql_exec_function(func, &fake_fcinfo, simple_eval_estate);
+		retval = plpgsql_exec_function(func, &fake_fcinfo, simple_eval_estate, codeblock->atomic);
 	}
 	PG_CATCH();
 	{
diff --git a/src/pl/plpgsql/src/pl_scanner.c b/src/pl/plpgsql/src/pl_scanner.c
index 65774f9902..08614a89a8 100644
--- a/src/pl/plpgsql/src/pl_scanner.c
+++ b/src/pl/plpgsql/src/pl_scanner.c
@@ -119,6 +119,7 @@ static const ScanKeyword unreserved_keywords[] = {
 	PG_KEYWORD("default", K_DEFAULT, UNRESERVED_KEYWORD)
 	PG_KEYWORD("detail", K_DETAIL, UNRESERVED_KEYWORD)
 	PG_KEYWORD("diagnostics", K_DIAGNOSTICS, UNRESERVED_KEYWORD)
+	PG_KEYWORD("do", K_DO, UNRESERVED_KEYWORD)
 	PG_KEYWORD("dump", K_DUMP, UNRESERVED_KEYWORD)
 	PG_KEYWORD("elseif", K_ELSIF, UNRESERVED_KEYWORD)
 	PG_KEYWORD("elsif", K_ELSIF, UNRESERVED_KEYWORD)
diff --git a/src/pl/plpgsql/src/plpgsql.h b/src/pl/plpgsql/src/plpgsql.h
index f7619a63f9..dc90fe532f 100644
--- a/src/pl/plpgsql/src/plpgsql.h
+++ b/src/pl/plpgsql/src/plpgsql.h
@@ -517,6 +517,7 @@ typedef struct PLpgSQL_stmt_call
 	PLpgSQL_stmt_type cmd_type;
 	int			lineno;
 	PLpgSQL_expr *expr;
+	bool		is_call;
 	PLpgSQL_variable *target;
 } PLpgSQL_stmt_call;
 
@@ -979,6 +980,7 @@ typedef struct PLpgSQL_execstate
 	bool		retisset;
 
 	bool		readonly_func;
+	bool		atomic;
 
 	char	   *exitlabel;		/* the "target" label of the current EXIT or
 								 * CONTINUE stmt, if any */
@@ -1194,7 +1196,8 @@ extern void _PG_init(void);
  */
 extern Datum plpgsql_exec_function(PLpgSQL_function *func,
 					  FunctionCallInfo fcinfo,
-					  EState *simple_eval_estate);
+					  EState *simple_eval_estate,
+					  bool atomic);
 extern HeapTuple plpgsql_exec_trigger(PLpgSQL_function *func,
 					 TriggerData *trigdata);
 extern void plpgsql_exec_event_trigger(PLpgSQL_function *func,
diff --git a/src/pl/plpgsql/src/sql/plpgsql_transaction.sql b/src/pl/plpgsql/src/sql/plpgsql_transaction.sql
index 02ee735079..3bbeab8978 100644
--- a/src/pl/plpgsql/src/sql/plpgsql_transaction.sql
+++ b/src/pl/plpgsql/src/sql/plpgsql_transaction.sql
@@ -1,12 +1,12 @@
 CREATE TABLE test1 (a int, b text);
 
 
-CREATE PROCEDURE transaction_test1()
+CREATE PROCEDURE transaction_test1(x int, y text)
 LANGUAGE plpgsql
 AS $$
 BEGIN
-    FOR i IN 0..9 LOOP
-        INSERT INTO test1 (a) VALUES (i);
+    FOR i IN 0..x LOOP
+        INSERT INTO test1 (a, b) VALUES (i, y);
         IF i % 2 = 0 THEN
             COMMIT;
         ELSE
@@ -16,7 +16,7 @@ CREATE PROCEDURE transaction_test1()
 END
 $$;
 
-CALL transaction_test1();
+CALL transaction_test1(9, 'foo');
 
 SELECT * FROM test1;
 
@@ -43,7 +43,7 @@ CREATE PROCEDURE transaction_test1()
 
 -- transaction commands not allowed when called in transaction block
 START TRANSACTION;
-CALL transaction_test1();
+CALL transaction_test1(9, 'error');
 COMMIT;
 
 START TRANSACTION;
@@ -80,7 +80,7 @@ CREATE FUNCTION transaction_test3() RETURNS int
 LANGUAGE plpgsql
 AS $$
 BEGIN
-    CALL transaction_test1();
+    CALL transaction_test1(9, 'error');
     RETURN 1;
 END;
 $$;
@@ -116,6 +116,36 @@ CREATE PROCEDURE transaction_test5()
 CALL transaction_test5();
 
 
+TRUNCATE test1;
+
+-- nested procedure calls
+CREATE PROCEDURE transaction_test6(c text)
+LANGUAGE plpgsql
+AS $$
+BEGIN
+    CALL transaction_test1(9, c);
+END;
+$$;
+
+CALL transaction_test6('bar');
+
+SELECT * FROM test1;
+
+TRUNCATE test1;
+
+CREATE PROCEDURE transaction_test7()
+LANGUAGE plpgsql
+AS $$
+BEGIN
+    DO 'BEGIN CALL transaction_test1(9, $x$baz$x$); END;';
+END;
+$$;
+
+CALL transaction_test7();
+
+SELECT * FROM test1;
+
+
 -- commit inside cursor loop
 CREATE TABLE test2 (x int);
 INSERT INTO test2 VALUES (0), (1), (2), (3), (4);

base-commit: 2cf8c7aa48559699f0607f5cb77b661156ad9750
-- 
2.16.2

From 48e703670579b5a5049a0638a156975178d71ed5 Mon Sep 17 00:00:00 2001
From: Peter Eisentraut <peter_e@gmx.net>
Date: Sat, 24 Mar 2018 10:05:06 -0400
Subject: [PATCH v3] PL/pgSQL: Nested CALL with transactions

So far, a nested CALL or DO in PL/pgSQL would not establish a context
where transaction control statements were allowed.  This fixes that by
handling CALL and DO specially in PL/pgSQL, passing the atomic/nonatomic
execution context through and doing the required management around
transaction boundaries.
---
 doc/src/sgml/plpgsql.sgml                          | 20 +++++-
 src/backend/executor/spi.c                         | 34 +++++----
 src/backend/tcop/utility.c                         |  2 +-
 src/include/executor/spi_priv.h                    |  1 +
 src/include/tcop/utility.h                         |  1 +
 .../plpgsql/src/expected/plpgsql_transaction.out   | 83 +++++++++++++++++-----
 src/pl/plpgsql/src/pl_exec.c                       | 69 +++++++++++++-----
 src/pl/plpgsql/src/pl_funcs.c                      |  4 +-
 src/pl/plpgsql/src/pl_gram.y                       | 18 +++++
 src/pl/plpgsql/src/pl_handler.c                    |  4 +-
 src/pl/plpgsql/src/pl_scanner.c                    |  1 +
 src/pl/plpgsql/src/plpgsql.h                       |  5 +-
 src/pl/plpgsql/src/sql/plpgsql_transaction.sql     | 52 ++++++++++++--
 13 files changed, 235 insertions(+), 59 deletions(-)

diff --git a/doc/src/sgml/plpgsql.sgml b/doc/src/sgml/plpgsql.sgml
index 7ed926fd51..b63b8496c7 100644
--- a/doc/src/sgml/plpgsql.sgml
+++ b/doc/src/sgml/plpgsql.sgml
@@ -3463,9 +3463,9 @@ <title>Looping Through a Cursor's Result</title>
    <title>Transaction Management</title>
 
    <para>
-    In procedures invoked by the <command>CALL</command> command from the top
-    level as well as in anonymous code blocks (<command>DO</command> command)
-    called from the top level, it is possible to end transactions using the
+    In procedures invoked by the <command>CALL</command> command
+    as well as in anonymous code blocks (<command>DO</command> command),
+    it is possible to end transactions using the
     commands <command>COMMIT</command> and <command>ROLLBACK</command>.  A new
     transaction is started automatically after a transaction is ended using
     these commands, so there is no separate <command>START
@@ -3495,6 +3495,20 @@ <title>Transaction Management</title>
 </programlisting>
    </para>
 
+   <para>
+    Transaction control is only possible in <command>CALL</command> or
+    <command>DO</command> invocations from the top level or nested
+    <command>CALL</command> or <command>DO</command> invocations without any
+    other intervening command.  For example, if the call stack is
+    <command>CALL proc1()</command> &rarr; <command>CALL proc2()</command>
+    &rarr; <command>CALL proc3()</command>, then the second and third
+    procedures can perform transaction control actions.  But if the call stack
+    is <command>CALL proc1()</command> &rarr; <command>SELECT
+    func2()</command> &rarr; <command>CALL proc3()</command>, then the last
+    procedure cannot do transaction control, because of the
+    <command>SELECT</command> in between.
+   </para>
+
    <para>
     A transaction cannot be ended inside a loop over a query result, nor
     inside a block with exception handlers.
diff --git a/src/backend/executor/spi.c b/src/backend/executor/spi.c
index 9fc4431b80..08f6f67a15 100644
--- a/src/backend/executor/spi.c
+++ b/src/backend/executor/spi.c
@@ -2041,8 +2041,11 @@ _SPI_execute_plan(SPIPlanPtr plan, ParamListInfo paramLI,
 	 *
 	 * In the first two cases, we can just push the snap onto the stack once
 	 * for the whole plan list.
+	 *
+	 * But if the plan has no_snapshots set to true, then don't manage
+	 * snapshots at all.  The caller should then take care of that.
 	 */
-	if (snapshot != InvalidSnapshot)
+	if (snapshot != InvalidSnapshot && !plan->no_snapshots)
 	{
 		if (read_only)
 		{
@@ -2121,7 +2124,7 @@ _SPI_execute_plan(SPIPlanPtr plan, ParamListInfo paramLI,
 		 * In the default non-read-only case, get a new snapshot, replacing
 		 * any that we pushed in a previous cycle.
 		 */
-		if (snapshot == InvalidSnapshot && !read_only)
+		if (snapshot == InvalidSnapshot && !read_only && !plan->no_snapshots)
 		{
 			if (pushed_active_snap)
 				PopActiveSnapshot();
@@ -2172,7 +2175,7 @@ _SPI_execute_plan(SPIPlanPtr plan, ParamListInfo paramLI,
 			 * If not read-only mode, advance the command counter before each
 			 * command and update the snapshot.
 			 */
-			if (!read_only)
+			if (!read_only && !plan->no_snapshots)
 			{
 				CommandCounterIncrement();
 				UpdateActiveSnapshotCommandId();
@@ -2203,10 +2206,23 @@ _SPI_execute_plan(SPIPlanPtr plan, ParamListInfo paramLI,
 			else
 			{
 				char		completionTag[COMPLETION_TAG_BUFSIZE];
+				ProcessUtilityContext context;
+
+				/*
+				 * If the SPI context is atomic, or we are asked to manage
+				 * snapshots, then we are in an atomic execution context.
+				 * Conversely, to propagate a nonatomic execution context, the
+				 * caller must be in a nonatomic SPI context and manage
+				 * snapshots itself.
+				 */
+				if (_SPI_current->atomic || !plan->no_snapshots)
+					context = PROCESS_UTILITY_QUERY;
+				else
+					context = PROCESS_UTILITY_QUERY_NONATOMIC;
 
 				ProcessUtility(stmt,
 							   plansource->query_string,
-							   PROCESS_UTILITY_QUERY,
+							   context,
 							   paramLI,
 							   _SPI_current->queryEnv,
 							   dest,
@@ -2638,11 +2654,8 @@ _SPI_make_plan_non_temp(SPIPlanPtr plan)
 	oldcxt = MemoryContextSwitchTo(plancxt);
 
 	/* Copy the SPI_plan struct and subsidiary data into the new context */
-	newplan = (SPIPlanPtr) palloc(sizeof(_SPI_plan));
+	newplan = (SPIPlanPtr) palloc0(sizeof(_SPI_plan));
 	newplan->magic = _SPI_PLAN_MAGIC;
-	newplan->saved = false;
-	newplan->oneshot = false;
-	newplan->plancache_list = NIL;
 	newplan->plancxt = plancxt;
 	newplan->cursor_options = plan->cursor_options;
 	newplan->nargs = plan->nargs;
@@ -2705,11 +2718,8 @@ _SPI_save_plan(SPIPlanPtr plan)
 	oldcxt = MemoryContextSwitchTo(plancxt);
 
 	/* Copy the SPI plan into its own context */
-	newplan = (SPIPlanPtr) palloc(sizeof(_SPI_plan));
+	newplan = (SPIPlanPtr) palloc0(sizeof(_SPI_plan));
 	newplan->magic = _SPI_PLAN_MAGIC;
-	newplan->saved = false;
-	newplan->oneshot = false;
-	newplan->plancache_list = NIL;
 	newplan->plancxt = plancxt;
 	newplan->cursor_options = plan->cursor_options;
 	newplan->nargs = plan->nargs;
diff --git a/src/backend/tcop/utility.c b/src/backend/tcop/utility.c
index e144583bd1..d355bef606 100644
--- a/src/backend/tcop/utility.c
+++ b/src/backend/tcop/utility.c
@@ -382,7 +382,7 @@ standard_ProcessUtility(PlannedStmt *pstmt,
 {
 	Node	   *parsetree = pstmt->utilityStmt;
 	bool		isTopLevel = (context == PROCESS_UTILITY_TOPLEVEL);
-	bool		isAtomicContext = (context != PROCESS_UTILITY_TOPLEVEL || IsTransactionBlock());
+	bool		isAtomicContext = (!(context == PROCESS_UTILITY_TOPLEVEL || context == PROCESS_UTILITY_QUERY_NONATOMIC) || IsTransactionBlock());
 	ParseState *pstate;
 
 	check_xact_readonly(parsetree);
diff --git a/src/include/executor/spi_priv.h b/src/include/executor/spi_priv.h
index 263c8f1453..376fae0bbc 100644
--- a/src/include/executor/spi_priv.h
+++ b/src/include/executor/spi_priv.h
@@ -86,6 +86,7 @@ typedef struct _SPI_plan
 	int			magic;			/* should equal _SPI_PLAN_MAGIC */
 	bool		saved;			/* saved or unsaved plan? */
 	bool		oneshot;		/* one-shot plan? */
+	bool		no_snapshots;	/* let the caller handle the snapshots */
 	List	   *plancache_list; /* one CachedPlanSource per parsetree */
 	MemoryContext plancxt;		/* Context containing _SPI_plan and data */
 	int			cursor_options; /* Cursor options used for planning */
diff --git a/src/include/tcop/utility.h b/src/include/tcop/utility.h
index 5550055710..880d19311a 100644
--- a/src/include/tcop/utility.h
+++ b/src/include/tcop/utility.h
@@ -20,6 +20,7 @@ typedef enum
 {
 	PROCESS_UTILITY_TOPLEVEL,	/* toplevel interactive command */
 	PROCESS_UTILITY_QUERY,		/* a complete query, but not toplevel */
+	PROCESS_UTILITY_QUERY_NONATOMIC, /* a complete query, nonatomic execution context */
 	PROCESS_UTILITY_SUBCOMMAND	/* a portion of a query */
 } ProcessUtilityContext;
 
diff --git a/src/pl/plpgsql/src/expected/plpgsql_transaction.out b/src/pl/plpgsql/src/expected/plpgsql_transaction.out
index ce66487137..b601f6aef6 100644
--- a/src/pl/plpgsql/src/expected/plpgsql_transaction.out
+++ b/src/pl/plpgsql/src/expected/plpgsql_transaction.out
@@ -1,10 +1,10 @@
 CREATE TABLE test1 (a int, b text);
-CREATE PROCEDURE transaction_test1()
+CREATE PROCEDURE transaction_test1(x int, y text)
 LANGUAGE plpgsql
 AS $$
 BEGIN
-    FOR i IN 0..9 LOOP
-        INSERT INTO test1 (a) VALUES (i);
+    FOR i IN 0..x LOOP
+        INSERT INTO test1 (a, b) VALUES (i, y);
         IF i % 2 = 0 THEN
             COMMIT;
         ELSE
@@ -13,15 +13,15 @@ BEGIN
     END LOOP;
 END
 $$;
-CALL transaction_test1();
+CALL transaction_test1(9, 'foo');
 SELECT * FROM test1;
- a | b 
----+---
- 0 | 
- 2 | 
- 4 | 
- 6 | 
- 8 | 
+ a |  b  
+---+-----
+ 0 | foo
+ 2 | foo
+ 4 | foo
+ 6 | foo
+ 8 | foo
 (5 rows)
 
 TRUNCATE test1;
@@ -51,9 +51,9 @@ SELECT * FROM test1;
 
 -- transaction commands not allowed when called in transaction block
 START TRANSACTION;
-CALL transaction_test1();
+CALL transaction_test1(9, 'error');
 ERROR:  invalid transaction termination
-CONTEXT:  PL/pgSQL function transaction_test1() line 6 at COMMIT
+CONTEXT:  PL/pgSQL function transaction_test1(integer,text) line 6 at COMMIT
 COMMIT;
 START TRANSACTION;
 DO LANGUAGE plpgsql $$ BEGIN COMMIT; END $$;
@@ -90,14 +90,14 @@ CREATE FUNCTION transaction_test3() RETURNS int
 LANGUAGE plpgsql
 AS $$
 BEGIN
-    CALL transaction_test1();
+    CALL transaction_test1(9, 'error');
     RETURN 1;
 END;
 $$;
 SELECT transaction_test3();
 ERROR:  invalid transaction termination
-CONTEXT:  PL/pgSQL function transaction_test1() line 6 at COMMIT
-SQL statement "CALL transaction_test1()"
+CONTEXT:  PL/pgSQL function transaction_test1(integer,text) line 6 at COMMIT
+SQL statement "CALL transaction_test1(9, 'error')"
 PL/pgSQL function transaction_test3() line 3 at CALL
 SELECT * FROM test1;
  a | b 
@@ -130,6 +130,57 @@ $$;
 CALL transaction_test5();
 ERROR:  invalid transaction termination
 CONTEXT:  PL/pgSQL function transaction_test5() line 3 at COMMIT
+TRUNCATE test1;
+-- nested procedure calls
+CREATE PROCEDURE transaction_test6(c text)
+LANGUAGE plpgsql
+AS $$
+BEGIN
+    CALL transaction_test1(9, c);
+END;
+$$;
+CALL transaction_test6('bar');
+SELECT * FROM test1;
+ a |  b  
+---+-----
+ 0 | bar
+ 2 | bar
+ 4 | bar
+ 6 | bar
+ 8 | bar
+(5 rows)
+
+TRUNCATE test1;
+CREATE PROCEDURE transaction_test7()
+LANGUAGE plpgsql
+AS $$
+BEGIN
+    DO 'BEGIN CALL transaction_test1(9, $x$baz$x$); END;';
+END;
+$$;
+CALL transaction_test7();
+SELECT * FROM test1;
+ a |  b  
+---+-----
+ 0 | baz
+ 2 | baz
+ 4 | baz
+ 6 | baz
+ 8 | baz
+(5 rows)
+
+CREATE PROCEDURE transaction_test8()
+LANGUAGE plpgsql
+AS $$
+BEGIN
+    EXECUTE 'CALL transaction_test1(10, $x$baz$x$)';
+END;
+$$;
+CALL transaction_test8();
+ERROR:  invalid transaction termination
+CONTEXT:  PL/pgSQL function transaction_test1(integer,text) line 6 at COMMIT
+SQL statement "CALL transaction_test1(10, $x$baz$x$)"
+PL/pgSQL function transaction_test8() line 3 at EXECUTE
 -- commit inside cursor loop
 CREATE TABLE test2 (x int);
 INSERT INTO test2 VALUES (0), (1), (2), (3), (4);
diff --git a/src/pl/plpgsql/src/pl_exec.c b/src/pl/plpgsql/src/pl_exec.c
index 38ea7a091f..fc0f0f0480 100644
--- a/src/pl/plpgsql/src/pl_exec.c
+++ b/src/pl/plpgsql/src/pl_exec.c
@@ -22,6 +22,7 @@
 #include "access/tupconvert.h"
 #include "catalog/pg_proc.h"
 #include "catalog/pg_type.h"
+#include "commands/defrem.h"
 #include "executor/execExpr.h"
 #include "executor/spi.h"
 #include "executor/spi_priv.h"
@@ -33,6 +34,7 @@
 #include "parser/scansup.h"
 #include "storage/proc.h"
 #include "tcop/tcopprot.h"
+#include "tcop/utility.h"
 #include "utils/array.h"
 #include "utils/builtins.h"
 #include "utils/datum.h"
@@ -311,7 +313,8 @@ static void plpgsql_estate_setup(PLpgSQL_execstate *estate,
 static void exec_eval_cleanup(PLpgSQL_execstate *estate);
 
 static void exec_prepare_plan(PLpgSQL_execstate *estate,
-				  PLpgSQL_expr *expr, int cursorOptions);
+							  PLpgSQL_expr *expr, int cursorOptions,
+							  bool keepplan);
 static void exec_simple_check_plan(PLpgSQL_execstate *estate, PLpgSQL_expr *expr);
 static void exec_save_simple_expr(PLpgSQL_expr *expr, CachedPlan *cplan);
 static void exec_check_rw_parameter(PLpgSQL_expr *expr, int target_dno);
@@ -440,7 +443,7 @@ static char *format_preparedparamsdata(PLpgSQL_execstate *estate,
  */
 Datum
 plpgsql_exec_function(PLpgSQL_function *func, FunctionCallInfo fcinfo,
-					  EState *simple_eval_estate)
+					  EState *simple_eval_estate, bool atomic)
 {
 	PLpgSQL_execstate estate;
 	ErrorContextCallback plerrcontext;
@@ -452,6 +455,7 @@ plpgsql_exec_function(PLpgSQL_function *func, FunctionCallInfo fcinfo,
 	 */
 	plpgsql_estate_setup(&estate, func, (ReturnSetInfo *) fcinfo->resultinfo,
 						 simple_eval_estate);
+	estate.atomic = atomic;
 
 	/*
 	 * Setup error traceback support for ereport()
@@ -2057,20 +2061,48 @@ exec_stmt_call(PLpgSQL_execstate *estate, PLpgSQL_stmt_call *stmt)
 {
 	PLpgSQL_expr *expr = stmt->expr;
 	ParamListInfo paramLI;
+	LocalTransactionId before_lxid;
+	LocalTransactionId after_lxid;
 	int			rc;
 
 	if (expr->plan == NULL)
-		exec_prepare_plan(estate, expr, 0);
+	{
+		/*
+		 * Don't save the plan if not in atomic context.  Otherwise,
+		 * transaction ends would cause warnings about plan leaks.
+		 */
+		exec_prepare_plan(estate, expr, 0, estate->atomic);
+
+		/*
+		 * The procedure call could end transactions, which would upset the
+		 * snapshot management in SPI_execute*, so don't let it do it.
+		 */
+		expr->plan->no_snapshots = true;
+	}
 
 	paramLI = setup_param_list(estate, expr);
 
+	before_lxid = MyProc->lxid;
+
 	rc = SPI_execute_plan_with_paramlist(expr->plan, paramLI,
 										 estate->readonly_func, 0);
 
+	after_lxid = MyProc->lxid;
+
 	if (rc < 0)
 		elog(ERROR, "SPI_execute_plan_with_paramlist failed executing query \"%s\": %s",
 			 expr->query, SPI_result_code_string(rc));
 
+	/*
+	 * If we are in a new transaction after the call, we need to reset some
+	 * internal state.
+	 */
+	if (before_lxid != after_lxid)
+	{
+		estate->simple_eval_estate = NULL;
+		plpgsql_create_econtext(estate);
+	}
+
 	if (SPI_processed == 1)
 	{
 		SPITupleTable *tuptab = SPI_tuptable;
@@ -2705,7 +2737,7 @@ exec_stmt_forc(PLpgSQL_execstate *estate, PLpgSQL_stmt_forc *stmt)
 	Assert(query);
 
 	if (query->plan == NULL)
-		exec_prepare_plan(estate, query, curvar->cursor_options);
+		exec_prepare_plan(estate, query, curvar->cursor_options, true);
 
 	/*
 	 * Set up ParamListInfo for this query
@@ -3719,6 +3751,7 @@ plpgsql_estate_setup(PLpgSQL_execstate *estate,
 	estate->retisset = func->fn_retset;
 
 	estate->readonly_func = func->fn_readonly;
+	estate->atomic = true;
 
 	estate->exitlabel = NULL;
 	estate->cur_error = NULL;
@@ -3863,7 +3896,8 @@ exec_eval_cleanup(PLpgSQL_execstate *estate)
  */
 static void
 exec_prepare_plan(PLpgSQL_execstate *estate,
-				  PLpgSQL_expr *expr, int cursorOptions)
+				  PLpgSQL_expr *expr, int cursorOptions,
+				  bool keepplan)
 {
 	SPIPlanPtr	plan;
 
@@ -3899,7 +3933,8 @@ exec_prepare_plan(PLpgSQL_execstate *estate,
 					 expr->query, SPI_result_code_string(SPI_result));
 		}
 	}
-	SPI_keepplan(plan);
+	if (keepplan)
+		SPI_keepplan(plan);
 	expr->plan = plan;
 
 	/* Check to see if it's a simple expression */
@@ -3938,7 +3973,7 @@ exec_stmt_execsql(PLpgSQL_execstate *estate,
 	{
 		ListCell   *l;
 
-		exec_prepare_plan(estate, expr, CURSOR_OPT_PARALLEL_OK);
+		exec_prepare_plan(estate, expr, CURSOR_OPT_PARALLEL_OK, true);
 		stmt->mod_stmt = false;
 		foreach(l, SPI_plan_get_plan_sources(expr->plan))
 		{
@@ -4396,7 +4431,7 @@ exec_stmt_open(PLpgSQL_execstate *estate, PLpgSQL_stmt_open *stmt)
 		 */
 		query = stmt->query;
 		if (query->plan == NULL)
-			exec_prepare_plan(estate, query, stmt->cursor_options);
+			exec_prepare_plan(estate, query, stmt->cursor_options, true);
 	}
 	else if (stmt->dynquery != NULL)
 	{
@@ -4467,7 +4502,7 @@ exec_stmt_open(PLpgSQL_execstate *estate, PLpgSQL_stmt_open *stmt)
 
 		query = curvar->cursor_explicit_expr;
 		if (query->plan == NULL)
-			exec_prepare_plan(estate, query, curvar->cursor_options);
+			exec_prepare_plan(estate, query, curvar->cursor_options, true);
 	}
 
 	/*
@@ -4707,7 +4742,7 @@ exec_assign_expr(PLpgSQL_execstate *estate, PLpgSQL_datum *target,
 	 */
 	if (expr->plan == NULL)
 	{
-		exec_prepare_plan(estate, expr, 0);
+		exec_prepare_plan(estate, expr, 0, true);
 		if (target->dtype == PLPGSQL_DTYPE_VAR)
 			exec_check_rw_parameter(expr, target->dno);
 	}
@@ -5566,7 +5601,7 @@ exec_eval_expr(PLpgSQL_execstate *estate,
 	 * If first time through, create a plan for this expression.
 	 */
 	if (expr->plan == NULL)
-		exec_prepare_plan(estate, expr, CURSOR_OPT_PARALLEL_OK);
+		exec_prepare_plan(estate, expr, CURSOR_OPT_PARALLEL_OK, true);
 
 	/*
 	 * If this is a simple expression, bypass SPI and use the executor
@@ -5652,7 +5687,7 @@ exec_run_select(PLpgSQL_execstate *estate,
 	 */
 	if (expr->plan == NULL)
 		exec_prepare_plan(estate, expr,
-						  portalP == NULL ? CURSOR_OPT_PARALLEL_OK : 0);
+						  portalP == NULL ? CURSOR_OPT_PARALLEL_OK : 0, true);
 
 	/*
 	 * Set up ParamListInfo to pass to executor
@@ -7834,11 +7869,13 @@ plpgsql_create_econtext(PLpgSQL_execstate *estate)
 	{
 		MemoryContext oldcontext;
 
-		Assert(shared_simple_eval_estate == NULL);
-		oldcontext = MemoryContextSwitchTo(TopTransactionContext);
-		shared_simple_eval_estate = CreateExecutorState();
+		if (shared_simple_eval_estate == NULL)
+		{
+			oldcontext = MemoryContextSwitchTo(TopTransactionContext);
+			shared_simple_eval_estate = CreateExecutorState();
+			MemoryContextSwitchTo(oldcontext);
+		}
 		estate->simple_eval_estate = shared_simple_eval_estate;
-		MemoryContextSwitchTo(oldcontext);
 	}
 
 	/*
diff --git a/src/pl/plpgsql/src/pl_funcs.c b/src/pl/plpgsql/src/pl_funcs.c
index 39d6a54663..fc96fb5f4d 100644
--- a/src/pl/plpgsql/src/pl_funcs.c
+++ b/src/pl/plpgsql/src/pl_funcs.c
@@ -285,7 +285,7 @@ plpgsql_stmt_typename(PLpgSQL_stmt *stmt)
 		case PLPGSQL_STMT_PERFORM:
 			return "PERFORM";
 		case PLPGSQL_STMT_CALL:
-			return "CALL";
+			return ((PLpgSQL_stmt_call *) stmt)->is_call ? "CALL" : "DO";
 		case PLPGSQL_STMT_COMMIT:
 			return "COMMIT";
 		case PLPGSQL_STMT_ROLLBACK:
@@ -1295,7 +1295,7 @@ static void
 dump_call(PLpgSQL_stmt_call *stmt)
 {
 	dump_ind();
-	printf("CALL expr = ");
+	printf("%s expr = ", stmt->is_call ? "CALL" : "DO");
 	dump_expr(stmt->expr);
 	printf("\n");
 }
diff --git a/src/pl/plpgsql/src/pl_gram.y b/src/pl/plpgsql/src/pl_gram.y
index 4c80936678..b8562ca8b4 100644
--- a/src/pl/plpgsql/src/pl_gram.y
+++ b/src/pl/plpgsql/src/pl_gram.y
@@ -276,6 +276,7 @@ static	void			check_raise_parameters(PLpgSQL_stmt_raise *stmt);
 %token <keyword>	K_DEFAULT
 %token <keyword>	K_DETAIL
 %token <keyword>	K_DIAGNOSTICS
+%token <keyword>	K_DO
 %token <keyword>	K_DUMP
 %token <keyword>	K_ELSE
 %token <keyword>	K_ELSIF
@@ -914,8 +915,24 @@ stmt_call		: K_CALL
 						new->cmd_type = PLPGSQL_STMT_CALL;
 						new->lineno = plpgsql_location_to_lineno(@1);
 						new->expr = read_sql_stmt("CALL ");
+						new->is_call = true;
 
 						$$ = (PLpgSQL_stmt *)new;
+
+					}
+				| K_DO
+					{
+						/* use the same structures as for CALL, for simplicity */
+						PLpgSQL_stmt_call *new;
+
+						new = palloc0(sizeof(PLpgSQL_stmt_call));
+						new->cmd_type = PLPGSQL_STMT_CALL;
+						new->lineno = plpgsql_location_to_lineno(@1);
+						new->expr = read_sql_stmt("DO ");
+						new->is_call = false;
+
+						$$ = (PLpgSQL_stmt *)new;
+
 					}
 				;
 
@@ -2434,6 +2451,7 @@ unreserved_keyword	:
 				| K_DEFAULT
 				| K_DETAIL
 				| K_DIAGNOSTICS
+				| K_DO
 				| K_DUMP
 				| K_ELSIF
 				| K_ERRCODE
diff --git a/src/pl/plpgsql/src/pl_handler.c b/src/pl/plpgsql/src/pl_handler.c
index f38ef04077..61452d9f7f 100644
--- a/src/pl/plpgsql/src/pl_handler.c
+++ b/src/pl/plpgsql/src/pl_handler.c
@@ -260,7 +260,7 @@ plpgsql_call_handler(PG_FUNCTION_ARGS)
 			retval = (Datum) 0;
 		}
 		else
-			retval = plpgsql_exec_function(func, fcinfo, NULL);
+			retval = plpgsql_exec_function(func, fcinfo, NULL, !nonatomic);
 	}
 	PG_CATCH();
 	{
@@ -332,7 +332,7 @@ plpgsql_inline_handler(PG_FUNCTION_ARGS)
 	/* And run the function */
 	PG_TRY();
 	{
-		retval = plpgsql_exec_function(func, &fake_fcinfo, simple_eval_estate);
+		retval = plpgsql_exec_function(func, &fake_fcinfo, simple_eval_estate, codeblock->atomic);
 	}
 	PG_CATCH();
 	{
diff --git a/src/pl/plpgsql/src/pl_scanner.c b/src/pl/plpgsql/src/pl_scanner.c
index 65774f9902..08614a89a8 100644
--- a/src/pl/plpgsql/src/pl_scanner.c
+++ b/src/pl/plpgsql/src/pl_scanner.c
@@ -119,6 +119,7 @@ static const ScanKeyword unreserved_keywords[] = {
 	PG_KEYWORD("default", K_DEFAULT, UNRESERVED_KEYWORD)
 	PG_KEYWORD("detail", K_DETAIL, UNRESERVED_KEYWORD)
 	PG_KEYWORD("diagnostics", K_DIAGNOSTICS, UNRESERVED_KEYWORD)
+	PG_KEYWORD("do", K_DO, UNRESERVED_KEYWORD)
 	PG_KEYWORD("dump", K_DUMP, UNRESERVED_KEYWORD)
 	PG_KEYWORD("elseif", K_ELSIF, UNRESERVED_KEYWORD)
 	PG_KEYWORD("elsif", K_ELSIF, UNRESERVED_KEYWORD)
diff --git a/src/pl/plpgsql/src/plpgsql.h b/src/pl/plpgsql/src/plpgsql.h
index f7619a63f9..dc90fe532f 100644
--- a/src/pl/plpgsql/src/plpgsql.h
+++ b/src/pl/plpgsql/src/plpgsql.h
@@ -517,6 +517,7 @@ typedef struct PLpgSQL_stmt_call
 	PLpgSQL_stmt_type cmd_type;
 	int			lineno;
 	PLpgSQL_expr *expr;
+	bool		is_call;
 	PLpgSQL_variable *target;
 } PLpgSQL_stmt_call;
 
@@ -979,6 +980,7 @@ typedef struct PLpgSQL_execstate
 	bool		retisset;
 
 	bool		readonly_func;
+	bool		atomic;
 
 	char	   *exitlabel;		/* the "target" label of the current EXIT or
 								 * CONTINUE stmt, if any */
@@ -1194,7 +1196,8 @@ extern void _PG_init(void);
  */
 extern Datum plpgsql_exec_function(PLpgSQL_function *func,
 					  FunctionCallInfo fcinfo,
-					  EState *simple_eval_estate);
+					  EState *simple_eval_estate,
+					  bool atomic);
 extern HeapTuple plpgsql_exec_trigger(PLpgSQL_function *func,
 					 TriggerData *trigdata);
 extern void plpgsql_exec_event_trigger(PLpgSQL_function *func,
diff --git a/src/pl/plpgsql/src/sql/plpgsql_transaction.sql b/src/pl/plpgsql/src/sql/plpgsql_transaction.sql
index 02ee735079..a718f50f89 100644
--- a/src/pl/plpgsql/src/sql/plpgsql_transaction.sql
+++ b/src/pl/plpgsql/src/sql/plpgsql_transaction.sql
@@ -1,12 +1,12 @@
 CREATE TABLE test1 (a int, b text);
 
 
-CREATE PROCEDURE transaction_test1()
+CREATE PROCEDURE transaction_test1(x int, y text)
 LANGUAGE plpgsql
 AS $$
 BEGIN
-    FOR i IN 0..9 LOOP
-        INSERT INTO test1 (a) VALUES (i);
+    FOR i IN 0..x LOOP
+        INSERT INTO test1 (a, b) VALUES (i, y);
         IF i % 2 = 0 THEN
             COMMIT;
         ELSE
@@ -16,7 +16,7 @@ CREATE PROCEDURE transaction_test1()
 END
 $$;
 
-CALL transaction_test1();
+CALL transaction_test1(9, 'foo');
 
 SELECT * FROM test1;
 
@@ -43,7 +43,7 @@ CREATE PROCEDURE transaction_test1()
 
 -- transaction commands not allowed when called in transaction block
 START TRANSACTION;
-CALL transaction_test1();
+CALL transaction_test1(9, 'error');
 COMMIT;
 
 START TRANSACTION;
@@ -80,7 +80,7 @@ CREATE FUNCTION transaction_test3() RETURNS int
 LANGUAGE plpgsql
 AS $$
 BEGIN
-    CALL transaction_test1();
+    CALL transaction_test1(9, 'error');
     RETURN 1;
 END;
 $$;
@@ -116,6 +116,46 @@ CREATE PROCEDURE transaction_test5()
 CALL transaction_test5();
 
 
+TRUNCATE test1;
+
+-- nested procedure calls
+CREATE PROCEDURE transaction_test6(c text)
+LANGUAGE plpgsql
+AS $$
+BEGIN
+    CALL transaction_test1(9, c);
+END;
+$$;
+
+CALL transaction_test6('bar');
+
+SELECT * FROM test1;
+
+TRUNCATE test1;
+
+CREATE PROCEDURE transaction_test7()
+LANGUAGE plpgsql
+AS $$
+BEGIN
+    DO 'BEGIN CALL transaction_test1(9, $x$baz$x$); END;';
+END;
+$$;
+
+CALL transaction_test7();
+
+SELECT * FROM test1;
+
+CREATE PROCEDURE transaction_test8()
+LANGUAGE plpgsql
+AS $$
+BEGIN
+    EXECUTE 'CALL transaction_test1(10, $x$baz$x$)';
+END;
+$$;
+
+CALL transaction_test8();
+
+
 -- commit inside cursor loop
 CREATE TABLE test2 (x int);
 INSERT INTO test2 VALUES (0), (1), (2), (3), (4);

base-commit: 52f3a9d6a32c0c070a15486c3aecbc4405d2da88
-- 
2.16.3