Add subtransactions for JsonExpr execution

Author: Nikita Glukhov

src/backend/executor/execExpr.c

@@ -2123,6 +2123,9 @@ ExecInitExprRec(Expr *node, PlanState *parent, ExprState *state,
 					}
 				}
 
+				if (jexpr->on_error.btype != JSON_BEHAVIOR_ERROR)
+					scratch.d.jsonexpr.volatility = getJsonExprVolatility(jexpr);
+
 				ExprEvalPushStep(state, &scratch);
 			}
 			break;

src/backend/executor/execExprInterp.c

@@ -59,6 +59,8 @@
 #include "postgres.h"
 
 #include "access/tuptoaster.h"
+#include "access/xact.h"
+#include "catalog/pg_proc.h"
 #include "catalog/pg_type.h"
 #include "commands/sequence.h"
 #include "executor/execExpr.h"
@@ -76,6 +78,7 @@
 #include "utils/jsonb.h"
 #include "utils/jsonpath.h"
 #include "utils/lsyscache.h"
+#include "utils/resowner.h"
 #include "utils/timestamp.h"
 #include "utils/typcache.h"
 #include "utils/xml.h"
@@ -3992,18 +3995,63 @@ ExecEvalJson(ExprState *state, ExprEvalStep *op, ExprContext *econtext)
 
 	if (jexpr->on_error.btype == JSON_BEHAVIOR_ERROR)
 	{
-		/* No need to use PG_TRY/PG_CATCH. */
+		/* No need to use PG_TRY/PG_CATCH with subtransactions. */
 		res = ExecEvalJsonExpr(state, op, econtext, jexpr, path, item, isjsonb,
 							   op->resnull);
 	}
 	else
 	{
+		/*
+		 * We should catch exceptions of category ERRCODE_DATA_EXCEPTION and
+		 * execute corresponding ON ERROR behavior.
+		 */
+		char		volatility = op->d.jsonexpr.volatility;
+		bool		useSubTransaction = volatility == PROVOLATILE_VOLATILE;
+		bool		useSubResourceOwner = volatility == PROVOLATILE_STABLE;
 		MemoryContext oldcontext = CurrentMemoryContext;
+		ResourceOwner oldowner = CurrentResourceOwner;
+
+		if (useSubTransaction)
+		{
+			BeginInternalSubTransaction(NULL);
+			/* Want to execute expressions inside function's memory context */
+			MemoryContextSwitchTo(oldcontext);
+		}
+		else if (useSubResourceOwner)
+		{
+			CurrentResourceOwner = ResourceOwnerCreate(CurrentResourceOwner,
+													   "JsonExpr");
+		}
 
 		PG_TRY();
 		{
-			res = ExecEvalJsonExpr(state, op, econtext, jexpr, path, item,
+			/*
+			 * We need to execute expressions with a new econtext
+			 * that belongs to the current subtransaction; if we try to use
+			 * the outer econtext then ExprContext shutdown callbacks will be
+			 * called at the wrong times.
+			 */
+			ExprContext *newecontext = useSubTransaction ?
+				CreateExprContext(econtext->ecxt_estate) : econtext;
+
+			res = ExecEvalJsonExpr(state, op, newecontext, jexpr, path, item,
 								   isjsonb, op->resnull);
+
+			if (useSubTransaction)
+			{
+				/* Commit the inner transaction, return to outer xact context */
+				ReleaseCurrentSubTransaction();
+				MemoryContextSwitchTo(oldcontext);
+				CurrentResourceOwner = oldowner;
+			}
+			else if (useSubResourceOwner)
+			{
+				/* buffer pins are released here: */
+				ResourceOwnerRelease(CurrentResourceOwner,
+									 RESOURCE_RELEASE_BEFORE_LOCKS,
+									 false, true);
+				CurrentResourceOwner = oldowner;
+			}
 		}
 		PG_CATCH();
 		{
@@ -4014,6 +4062,22 @@ ExecEvalJson(ExprState *state, ExprEvalStep *op, ExprContext *econtext)
 			edata = CopyErrorData();
 			FlushErrorState();
 
+			if (useSubTransaction)
+			{
+				/* Abort the inner transaction */
+				RollbackAndReleaseCurrentSubTransaction();
+				MemoryContextSwitchTo(oldcontext);
+				CurrentResourceOwner = oldowner;
+			}
+			else if (useSubResourceOwner)
+			{
+				/* buffer pins are released here: */
+				ResourceOwnerRelease(CurrentResourceOwner,
+									 RESOURCE_RELEASE_BEFORE_LOCKS,
+									 false, true);
+				CurrentResourceOwner = oldowner;
+			}
+
 			if (ERRCODE_TO_CATEGORY(edata->sqlerrcode) != ERRCODE_DATA_EXCEPTION)
 				ReThrowError(edata);
 

src/backend/parser/parse_expr.c

@@ -22,6 +22,7 @@
 #include "miscadmin.h"
 #include "nodes/makefuncs.h"
 #include "nodes/nodeFuncs.h"
+#include "optimizer/clauses.h"
 #include "optimizer/tlist.h"
 #include "optimizer/var.h"
 #include "parser/analyze.h"
@@ -4717,3 +4718,105 @@ transformJsonFuncExpr(ParseState *pstate, JsonFuncExpr *func)
 
 	return (Node *) jsexpr;
 }
+
+static char
+getExprVolatility(Node *expr)
+{
+	if (contain_volatile_functions(expr))
+		return PROVOLATILE_VOLATILE;
+
+	if (contain_mutable_functions(expr))
+		return PROVOLATILE_STABLE;
+
+	return PROVOLATILE_IMMUTABLE;
+}
+
+static char
+getJsonCoercionVolatility(JsonCoercion *coercion, JsonReturning *returning)
+{
+	if (!coercion)
+		return PROVOLATILE_IMMUTABLE;
+
+	if (coercion->expr)
+		return getExprVolatility(coercion->expr);
+
+	if (coercion->via_io)
+	{
+		Oid			typinput;
+		Oid			typioparam;
+
+		getTypeInputInfo(returning->typid, &typinput, &typioparam);
+
+		return func_volatile(typinput);
+	}
+
+	if (coercion->via_populate)
+		return PROVOLATILE_STABLE;
+
+	return PROVOLATILE_IMMUTABLE;
+}
+
+char
+getJsonExprVolatility(JsonExpr *jsexpr)
+{
+	char		volatility;
+	char		volatility2;
+	ListCell   *lc;
+
+	volatility = getExprVolatility(jsexpr->formatted_expr);
+
+	if (volatility == PROVOLATILE_VOLATILE)
+		return PROVOLATILE_VOLATILE;
+
+	volatility2 = getJsonCoercionVolatility(jsexpr->result_coercion,
+											&jsexpr->returning);
+
+	if (volatility2 == PROVOLATILE_VOLATILE)
+		return PROVOLATILE_VOLATILE;
+
+	if (volatility2 == PROVOLATILE_STABLE)
+		volatility = PROVOLATILE_STABLE;
+
+	if (jsexpr->coercions)
+	{
+		JsonCoercion **coercion;
+
+		for (coercion = &jsexpr->coercions->null;
+			 coercion <= &jsexpr->coercions->composite;
+			 coercion++)
+		{
+			volatility2 = getJsonCoercionVolatility(*coercion, &jsexpr->returning);
+
+			if (volatility2 == PROVOLATILE_VOLATILE)
+				return PROVOLATILE_VOLATILE;
+
+			if (volatility2 == PROVOLATILE_STABLE)
+				volatility = PROVOLATILE_STABLE;
+		}
+	}
+
+	if (jsexpr->on_empty.btype == JSON_BEHAVIOR_DEFAULT)
+	{
+		volatility2 = getExprVolatility(jsexpr->on_empty.default_expr);
+
+		if (volatility2 == PROVOLATILE_VOLATILE)
+			return PROVOLATILE_VOLATILE;
+
+		if (volatility2 == PROVOLATILE_STABLE)
+			volatility = PROVOLATILE_STABLE;
+	}
+
+	foreach(lc, jsexpr->passing.values)
+	{
+		volatility2 = getExprVolatility(lfirst(lc));
+
+		if (volatility2 == PROVOLATILE_VOLATILE)
+			return PROVOLATILE_VOLATILE;
+
+		if (volatility2 == PROVOLATILE_STABLE)
+			volatility = PROVOLATILE_STABLE;
+	}
+
+	return volatility;
+}
+

src/include/executor/execExpr.h

@@ -567,6 +567,7 @@ typedef struct ExprEvalStep
 		struct
 		{
 			JsonExpr   *jsexpr;			/* original expression node */
+			char		volatility;		/* volatility of subexpressions */
 
 			struct
 			{