plan cache overhead on plpgsql expression

diff --git a/src/backend/commands/tablecmds.c b/src/backend/commands/tablecmds.c
index b7c8d663fc..b292948853 100644
--- a/src/backend/commands/tablecmds.c
+++ b/src/backend/commands/tablecmds.c
@@ -16539,9 +16539,10 @@ CloneRowTriggersToPartition(Relation parent, Relation partition)
 		 * Internal triggers require careful examination.  Ideally, we don't
 		 * clone them.
 		 *
-		 * However, if our parent is a partitioned relation, there might be
-		 * internal triggers that need cloning.  In that case, we must skip
-		 * clone it if the trigger on parent depends on another trigger.
+		 * However, if our parent is a partition itself, there might be
+		 * internal triggers that need cloning.  For example, triggers on the
+		 * parent that were in turn cloned from its own parent are marked
+		 * internal, which too must be cloned to the partition.
 		 *
 		 * Note we dare not verify that the other trigger belongs to an
 		 * ancestor relation of our parent, because that creates deadlock
diff --git a/src/pl/plpgsql/src/pl_exec.c b/src/pl/plpgsql/src/pl_exec.c
index 5acf604f63..72f156726f 100644
--- a/src/pl/plpgsql/src/pl_exec.c
+++ b/src/pl/plpgsql/src/pl_exec.c
@@ -319,8 +319,7 @@ static void exec_eval_cleanup(PLpgSQL_execstate *estate);
 static void exec_prepare_plan(PLpgSQL_execstate *estate,
 							  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_simple_expr(PLpgSQL_execstate *estate, PLpgSQL_expr *expr);
 static void exec_check_rw_parameter(PLpgSQL_expr *expr, int target_dno);
 static bool contains_target_param(Node *node, int *target_dno);
 static bool exec_eval_simple_expr(PLpgSQL_execstate *estate,
@@ -4062,7 +4061,7 @@ exec_prepare_plan(PLpgSQL_execstate *estate,
 	expr->plan = plan;
 
 	/* Check to see if it's a simple expression */
-	exec_simple_check_plan(estate, expr);
+	exec_check_simple_expr(estate, expr);
 
 	/*
 	 * Mark expression as not using a read-write param.  exec_assign_value has
@@ -5758,9 +5757,10 @@ exec_eval_expr(PLpgSQL_execstate *estate,
 	Form_pg_attribute attr;
 
 	/*
-	 * If first time through, create a plan for this expression.
+	 * Create a plan for this expression, if first time through or the
+	 * existing plan is no longer valid.
 	 */
-	if (expr->plan == NULL)
+	if (expr->plan == NULL || !SPI_plan_is_valid(expr->plan))
 		exec_prepare_plan(estate, expr, CURSOR_OPT_PARALLEL_OK, true);
 
 	/*
@@ -6076,7 +6076,6 @@ exec_eval_simple_expr(PLpgSQL_execstate *estate,
 {
 	ExprContext *econtext = estate->eval_econtext;
 	LocalTransactionId curlxid = MyProc->lxid;
-	CachedPlan *cplan;
 	void	   *save_setup_arg;
 	bool		need_snapshot;
 	MemoryContext oldcontext;
@@ -6093,30 +6092,15 @@ exec_eval_simple_expr(PLpgSQL_execstate *estate,
 	if (expr->expr_simple_in_use && expr->expr_simple_lxid == curlxid)
 		return false;
 
-	/*
-	 * Revalidate cached plan, so that we will notice if it became stale. (We
-	 * need to hold a refcount while using the plan, anyway.)  If replanning
-	 * is needed, do that work in the eval_mcontext.
-	 */
-	oldcontext = MemoryContextSwitchTo(get_eval_mcontext(estate));
-	cplan = SPI_plan_get_cached_plan(expr->plan);
-	MemoryContextSwitchTo(oldcontext);
+	/* Simple expressions aren't planned, so generation wouldn't change. */
+	Assert(expr->expr_simple_generation == 0);
 
 	/*
-	 * We can't get a failure here, because the number of CachedPlanSources in
-	 * the SPI plan can't change from what exec_simple_check_plan saw; it's a
-	 * property of the raw parsetree generated from the query text.
+	 * better recheck r/w safety, as it could change due to inlining
+	 * XXX - should no longer occur, because planning is not done?
 	 */
-	Assert(cplan != NULL);
-
-	/* If it got replanned, update our copy of the simple expression */
-	if (cplan->generation != expr->expr_simple_generation)
-	{
-		exec_save_simple_expr(expr, cplan);
-		/* better recheck r/w safety, as it could change due to inlining */
-		if (expr->rwparam >= 0)
-			exec_check_rw_parameter(expr, expr->rwparam);
-	}
+	if (expr->rwparam >= 0)
+		exec_check_rw_parameter(expr, expr->rwparam);
 
 	/*
 	 * Pass back previously-determined result type.
@@ -6192,11 +6176,6 @@ exec_eval_simple_expr(PLpgSQL_execstate *estate,
 
 	MemoryContextSwitchTo(oldcontext);
 
-	/*
-	 * Now we can release our refcount on the cached plan.
-	 */
-	ReleaseCachedPlan(cplan, true);
-
 	/*
 	 * That's it.
 	 */
@@ -7890,19 +7869,20 @@ get_cast_hashentry(PLpgSQL_execstate *estate,
 
 
 /* ----------
- * exec_simple_check_plan -		Check if a plan is simple enough to
+ * exec_check_simple_expr -		Check if the expression is simple enough to
  *								be evaluated by ExecEvalExpr() instead
  *								of SPI.
+ *
+ * If it is, set expr->expr_simple_expr.
  * ----------
  */
 static void
-exec_simple_check_plan(PLpgSQL_execstate *estate, PLpgSQL_expr *expr)
+exec_check_simple_expr(PLpgSQL_execstate *estate, PLpgSQL_expr *expr)
 {
 	List	   *plansources;
 	CachedPlanSource *plansource;
 	Query	   *query;
-	CachedPlan *cplan;
-	MemoryContext oldcontext;
+	Expr	   *tle_expr;
 
 	/*
 	 * Initialize to "not simple".
@@ -7972,95 +7952,12 @@ exec_simple_check_plan(PLpgSQL_execstate *estate, PLpgSQL_expr *expr)
 		return;
 
 	/*
-	 * OK, we can treat it as a simple plan.
-	 *
-	 * Get the generic plan for the query.  If replanning is needed, do that
-	 * work in the eval_mcontext.
-	 */
-	oldcontext = MemoryContextSwitchTo(get_eval_mcontext(estate));
-	cplan = SPI_plan_get_cached_plan(expr->plan);
-	MemoryContextSwitchTo(oldcontext);
-
-	/* Can't fail, because we checked for a single CachedPlanSource above */
-	Assert(cplan != NULL);
-
-	/* Share the remaining work with replan code path */
-	exec_save_simple_expr(expr, cplan);
-
-	/* Release our plan refcount */
-	ReleaseCachedPlan(cplan, true);
-}
-
-/*
- * exec_save_simple_expr --- extract simple expression from CachedPlan
- */
-static void
-exec_save_simple_expr(PLpgSQL_expr *expr, CachedPlan *cplan)
-{
-	PlannedStmt *stmt;
-	Plan	   *plan;
-	Expr	   *tle_expr;
-
-	/*
-	 * Given the checks that exec_simple_check_plan did, none of the Asserts
-	 * here should ever fail.
-	 */
-
-	/* Extract the single PlannedStmt */
-	Assert(list_length(cplan->stmt_list) == 1);
-	stmt = linitial_node(PlannedStmt, cplan->stmt_list);
-	Assert(stmt->commandType == CMD_SELECT);
-
-	/*
-	 * Ordinarily, the plan node should be a simple Result.  However, if
-	 * force_parallel_mode is on, the planner might've stuck a Gather node
-	 * atop that.  The simplest way to deal with this is to look through the
-	 * Gather node.  The Gather node's tlist would normally contain a Var
-	 * referencing the child node's output, but it could also be a Param, or
-	 * it could be a Const that setrefs.c copied as-is.
-	 */
-	plan = stmt->planTree;
-	for (;;)
-	{
-		/* Extract the single tlist expression */
-		Assert(list_length(plan->targetlist) == 1);
-		tle_expr = castNode(TargetEntry, linitial(plan->targetlist))->expr;
-
-		if (IsA(plan, Result))
-		{
-			Assert(plan->lefttree == NULL &&
-				   plan->righttree == NULL &&
-				   plan->initPlan == NULL &&
-				   plan->qual == NULL &&
-				   ((Result *) plan)->resconstantqual == NULL);
-			break;
-		}
-		else if (IsA(plan, Gather))
-		{
-			Assert(plan->lefttree != NULL &&
-				   plan->righttree == NULL &&
-				   plan->initPlan == NULL &&
-				   plan->qual == NULL);
-			/* If setrefs.c copied up a Const, no need to look further */
-			if (IsA(tle_expr, Const))
-				break;
-			/* Otherwise, it had better be a Param or an outer Var */
-			Assert(IsA(tle_expr, Param) ||(IsA(tle_expr, Var) &&
-										   ((Var *) tle_expr)->varno == OUTER_VAR));
-			/* Descend to the child node */
-			plan = plan->lefttree;
-		}
-		else
-			elog(ERROR, "unexpected plan node type: %d",
-				 (int) nodeTag(plan));
-	}
-
-	/*
-	 * Save the simple expression, and initialize state to "not valid in
-	 * current transaction".
+	 * We have a simple expression.  Save it and initialize state to "not
+	 * valid in current transaction".
 	 */
+	tle_expr = castNode(TargetEntry, linitial(query->targetList))->expr;
 	expr->expr_simple_expr = tle_expr;
-	expr->expr_simple_generation = cplan->generation;
+	expr->expr_simple_generation = plansource->generation;
 	expr->expr_simple_state = NULL;
 	expr->expr_simple_in_use = false;
 	expr->expr_simple_lxid = InvalidLocalTransactionId;
diff --git a/src/test/regress/expected/plpgsql.out b/src/test/regress/expected/plpgsql.out
index cd2c79f4d5..3abbc7bc57 100644
--- a/src/test/regress/expected/plpgsql.out
+++ b/src/test/regress/expected/plpgsql.out
@@ -4289,12 +4289,10 @@ end
 $$;
 select fail();
 ERROR:  division by zero
-CONTEXT:  SQL statement "SELECT 1/0"
-PL/pgSQL function fail() line 3 at RETURN
+CONTEXT:  PL/pgSQL function fail() line 3 at RETURN
 select fail();
 ERROR:  division by zero
-CONTEXT:  SQL statement "SELECT 1/0"
-PL/pgSQL function fail() line 3 at RETURN
+CONTEXT:  PL/pgSQL function fail() line 3 at RETURN
 drop function fail();
 -- Test handling of string literals.
 set standard_conforming_strings = off;

diff --git a/src/backend/optimizer/util/clauses.c b/src/backend/optimizer/util/clauses.c
index 2d3ec22407..5ce0079a12 100644
--- a/src/backend/optimizer/util/clauses.c
+++ b/src/backend/optimizer/util/clauses.c
@@ -108,6 +108,7 @@ static bool contain_volatile_functions_not_nextval_walker(Node *node, void *cont
 static bool max_parallel_hazard_walker(Node *node,
 									   max_parallel_hazard_context *context);
 static bool contain_nonstrict_functions_walker(Node *node, void *context);
+static bool contain_inlinable_functions_walker(Node *node, void *context);
 static bool contain_context_dependent_node(Node *clause);
 static bool contain_context_dependent_node_walker(Node *node, int *flags);
 static bool contain_leaked_vars_walker(Node *node, void *context);
@@ -1218,6 +1219,63 @@ contain_nonstrict_functions_walker(Node *node, void *context)
 								  context);
 }
 
+/*****************************************************************************
+ *		Check clauses for inline-able functions
+ *****************************************************************************/
+
+bool
+contain_inlinable_functions(Node *node)
+{
+	return contain_inlinable_functions_walker(node, NULL);
+}
+
+/*
+ * can_inline_function - checks if a function is inline-able
+ */
+static bool
+can_inline_function(HeapTuple func_tuple)
+{
+	Form_pg_proc funcform = (Form_pg_proc) GETSTRUCT(func_tuple);
+
+	/*
+	 * Nope if the function is not SQL-language or has other showstopper
+	 * properties.  (The prokind and nargs checks are just paranoia.)
+	 */
+	return	funcform->prolang == SQLlanguageId &&
+			funcform->prokind == PROKIND_FUNCTION &&
+			!funcform->prosecdef && !funcform->proretset &&
+			funcform->prorettype != RECORDOID &&
+			heap_attisnull(func_tuple, Anum_pg_proc_proconfig, NULL);
+}
+
+static bool
+can_inline_function_checker(Oid funcid, void *context)
+{
+	HeapTuple	func_tuple;
+	bool		result;
+
+	func_tuple = SearchSysCache1(PROCOID, ObjectIdGetDatum(funcid));
+	if (!HeapTupleIsValid(func_tuple))
+		elog(ERROR, "cache lookup failed for function %u", funcid);
+
+	result = can_inline_function(func_tuple);
+	ReleaseSysCache(func_tuple);
+
+	return result;
+}
+
+static bool
+contain_inlinable_functions_walker(Node *node, void *context)
+{
+	if (node == NULL)
+		return false;
+	if (check_functions_in_node(node, can_inline_function_checker, context))
+		return true;
+
+	return expression_tree_walker(node, contain_inlinable_functions_walker,
+								  context);
+}
+
 /*****************************************************************************
  *		Check clauses for context-dependent nodes
  *****************************************************************************/
@@ -4022,7 +4080,8 @@ simplify_function(Oid funcid, Oid result_type, int32 result_typmod,
 		Assert(newexpr != (Expr *) &fexpr);
 	}
 
-	if (!newexpr && allow_non_const)
+	if (!newexpr && allow_non_const &&
+		can_inline_function(func_tuple))
 		newexpr = inline_function(funcid, result_type, result_collid,
 								  input_collid, args, funcvariadic,
 								  func_tuple, context);
@@ -4415,16 +4474,11 @@ inline_function(Oid funcid, Oid result_type, Oid result_collid,
 	int			i;
 
 	/*
-	 * Forget it if the function is not SQL-language or has other showstopper
-	 * properties.  (The prokind and nargs checks are just paranoia.)
+	 * Caller should already have checked whether the function can be inlined
+	 * using can_function_inline().
 	 */
-	if (funcform->prolang != SQLlanguageId ||
-		funcform->prokind != PROKIND_FUNCTION ||
-		funcform->prosecdef ||
-		funcform->proretset ||
-		funcform->prorettype == RECORDOID ||
-		!heap_attisnull(func_tuple, Anum_pg_proc_proconfig, NULL) ||
-		funcform->pronargs != list_length(args))
+
+	if (funcform->pronargs != list_length(args))
 		return NULL;
 
 	/* Check for recursive function, and give up trying to expand if so */
diff --git a/src/include/optimizer/optimizer.h b/src/include/optimizer/optimizer.h
index 5283995df8..4954657147 100644
--- a/src/include/optimizer/optimizer.h
+++ b/src/include/optimizer/optimizer.h
@@ -130,6 +130,7 @@ extern Expr *canonicalize_qual(Expr *qual, bool is_check);
 extern bool contain_mutable_functions(Node *clause);
 extern bool contain_volatile_functions(Node *clause);
 extern bool contain_volatile_functions_not_nextval(Node *clause);
+extern bool contain_inlinable_functions(Node *node);
 
 extern Node *eval_const_expressions(PlannerInfo *root, Node *node);
 
diff --git a/src/pl/plpgsql/src/pl_exec.c b/src/pl/plpgsql/src/pl_exec.c
index 5acf604f63..244f7b576f 100644
--- a/src/pl/plpgsql/src/pl_exec.c
+++ b/src/pl/plpgsql/src/pl_exec.c
@@ -5758,9 +5758,10 @@ exec_eval_expr(PLpgSQL_execstate *estate,
 	Form_pg_attribute attr;
 
 	/*
-	 * If first time through, create a plan for this expression.
+	 * Create a plan for this expression, if first time through or the
+	 * existing plan is no longer valid.
 	 */
-	if (expr->plan == NULL)
+	if (expr->plan == NULL || !SPI_plan_is_valid(expr->plan))
 		exec_prepare_plan(estate, expr, CURSOR_OPT_PARALLEL_OK, true);
 
 	/*
@@ -6076,7 +6077,6 @@ exec_eval_simple_expr(PLpgSQL_execstate *estate,
 {
 	ExprContext *econtext = estate->eval_econtext;
 	LocalTransactionId curlxid = MyProc->lxid;
-	CachedPlan *cplan;
 	void	   *save_setup_arg;
 	bool		need_snapshot;
 	MemoryContext oldcontext;
@@ -6093,29 +6093,39 @@ exec_eval_simple_expr(PLpgSQL_execstate *estate,
 	if (expr->expr_simple_in_use && expr->expr_simple_lxid == curlxid)
 		return false;
 
-	/*
-	 * Revalidate cached plan, so that we will notice if it became stale. (We
-	 * need to hold a refcount while using the plan, anyway.)  If replanning
-	 * is needed, do that work in the eval_mcontext.
-	 */
-	oldcontext = MemoryContextSwitchTo(get_eval_mcontext(estate));
-	cplan = SPI_plan_get_cached_plan(expr->plan);
-	MemoryContextSwitchTo(oldcontext);
+	if (expr->expr_contains_inline_func)
+	{
+		CachedPlan *cplan;
 
-	/*
-	 * We can't get a failure here, because the number of CachedPlanSources in
-	 * the SPI plan can't change from what exec_simple_check_plan saw; it's a
-	 * property of the raw parsetree generated from the query text.
-	 */
-	Assert(cplan != NULL);
+		/*
+		 * Revalidate cached plan, so that we will notice if it became stale.
+		 * (We need to hold a refcount while using the plan, anyway.)  If
+		 * replanning is needed, do that work in the eval_mcontext.
+		 */
+		oldcontext = MemoryContextSwitchTo(get_eval_mcontext(estate));
+		cplan = SPI_plan_get_cached_plan(expr->plan);
+		MemoryContextSwitchTo(oldcontext);
 
-	/* If it got replanned, update our copy of the simple expression */
-	if (cplan->generation != expr->expr_simple_generation)
-	{
-		exec_save_simple_expr(expr, cplan);
-		/* better recheck r/w safety, as it could change due to inlining */
-		if (expr->rwparam >= 0)
-			exec_check_rw_parameter(expr, expr->rwparam);
+		/*
+		 * We can't get a failure here, because the number of CachedPlanSources
+		 * in the SPI plan can't change from what exec_simple_check_plan saw;
+		 * it's a property of the raw parsetree generated from the query text.
+		 */
+		Assert(cplan != NULL);
+
+		/* If it got replanned, update our copy of the simple expression */
+		if (cplan->generation != expr->expr_simple_generation)
+		{
+			exec_save_simple_expr(expr, cplan);
+			/* better recheck r/w safety, as it could change due to inlining */
+			if (expr->rwparam >= 0)
+				exec_check_rw_parameter(expr, expr->rwparam);
+		}
+
+		/*
+		 * Now we can release our refcount on the cached plan.
+		 */
+		ReleaseCachedPlan(cplan, true);
 	}
 
 	/*
@@ -6192,11 +6202,6 @@ exec_eval_simple_expr(PLpgSQL_execstate *estate,
 
 	MemoryContextSwitchTo(oldcontext);
 
-	/*
-	 * Now we can release our refcount on the cached plan.
-	 */
-	ReleaseCachedPlan(cplan, true);
-
 	/*
 	 * That's it.
 	 */
@@ -7890,9 +7895,11 @@ get_cast_hashentry(PLpgSQL_execstate *estate,
 
 
 /* ----------
- * exec_simple_check_plan -		Check if a plan is simple enough to
+ * exec_simple_check_plan -		Check if the expression is simple enough to
  *								be evaluated by ExecEvalExpr() instead
  *								of SPI.
+ *
+ * If it is, set expr->expr_simple_expr.
  * ----------
  */
 static void
@@ -7901,8 +7908,7 @@ exec_simple_check_plan(PLpgSQL_execstate *estate, PLpgSQL_expr *expr)
 	List	   *plansources;
 	CachedPlanSource *plansource;
 	Query	   *query;
-	CachedPlan *cplan;
-	MemoryContext oldcontext;
+	Expr	   *tle_expr;
 
 	/*
 	 * Initialize to "not simple".
@@ -7972,23 +7978,55 @@ exec_simple_check_plan(PLpgSQL_execstate *estate, PLpgSQL_expr *expr)
 		return;
 
 	/*
-	 * OK, we can treat it as a simple plan.
-	 *
-	 * Get the generic plan for the query.  If replanning is needed, do that
-	 * work in the eval_mcontext.
+	 * We have a simple expression.  Save it and initialize state to "not
+	 * valid in current transaction".
 	 */
-	oldcontext = MemoryContextSwitchTo(get_eval_mcontext(estate));
-	cplan = SPI_plan_get_cached_plan(expr->plan);
-	MemoryContextSwitchTo(oldcontext);
+	tle_expr = castNode(TargetEntry, linitial(query->targetList))->expr;
 
-	/* Can't fail, because we checked for a single CachedPlanSource above */
-	Assert(cplan != NULL);
+	if (contain_inlinable_functions((Node *) tle_expr))
+	{
+		CachedPlan *cplan = NULL;
+		MemoryContext oldcontext;
+
+		expr->expr_contains_inline_func = true;
 
-	/* Share the remaining work with replan code path */
-	exec_save_simple_expr(expr, cplan);
+		/*
+		 * Get the generic plan for the query.  If replanning is needed, do
+		 * that work in the eval_mcontext.
+		 */
+		oldcontext = MemoryContextSwitchTo(get_eval_mcontext(estate));
+		cplan = SPI_plan_get_cached_plan(expr->plan);
+		MemoryContextSwitchTo(oldcontext);
 
-	/* Release our plan refcount */
-	ReleaseCachedPlan(cplan, true);
+		/*
+		 * Can't fail, because we checked for a single CachedPlanSource
+		 * above.
+		 */
+		Assert(cplan != NULL);
+
+		exec_save_simple_expr(expr, cplan);
+
+		/* Release our plan refcount */
+		ReleaseCachedPlan(cplan, true);
+	}
+	else
+	{
+		/*
+		 * Save the simple expression, and initialize state to "not valid in
+		 * current transaction".
+		 */
+		expr->expr_simple_expr = tle_expr;
+		expr->expr_simple_generation = plansource->generation;
+		expr->expr_simple_state = NULL;
+		expr->expr_simple_in_use = false;
+		expr->expr_simple_lxid = InvalidLocalTransactionId;
+		/* Also stash away the expression result type */
+		expr->expr_simple_type = exprType((Node *) tle_expr);
+		expr->expr_simple_typmod = exprTypmod((Node *) tle_expr);
+		/* We also want to remember if it is immutable or not */
+		expr->expr_simple_mutable = contain_mutable_functions((Node *) tle_expr);
+		expr->expr_contains_inline_func = false;
+	}
 }
 
 /*
diff --git a/src/pl/plpgsql/src/plpgsql.h b/src/pl/plpgsql/src/plpgsql.h
index b599f67fc5..0e02212135 100644
--- a/src/pl/plpgsql/src/plpgsql.h
+++ b/src/pl/plpgsql/src/plpgsql.h
@@ -245,6 +245,7 @@ typedef struct PLpgSQL_expr
 	ExprState  *expr_simple_state;	/* eval tree for expr_simple_expr */
 	bool		expr_simple_in_use; /* true if eval tree is active */
 	LocalTransactionId expr_simple_lxid;
+	bool		expr_contains_inline_func;
 } PLpgSQL_expr;
 
 /*
diff --git a/src/test/regress/expected/plpgsql.out b/src/test/regress/expected/plpgsql.out
index cd2c79f4d5..3abbc7bc57 100644
--- a/src/test/regress/expected/plpgsql.out
+++ b/src/test/regress/expected/plpgsql.out
@@ -4289,12 +4289,10 @@ end
 $$;
 select fail();
 ERROR:  division by zero
-CONTEXT:  SQL statement "SELECT 1/0"
-PL/pgSQL function fail() line 3 at RETURN
+CONTEXT:  PL/pgSQL function fail() line 3 at RETURN
 select fail();
 ERROR:  division by zero
-CONTEXT:  SQL statement "SELECT 1/0"
-PL/pgSQL function fail() line 3 at RETURN
+CONTEXT:  PL/pgSQL function fail() line 3 at RETURN
 drop function fail();
 -- Test handling of string literals.
 set standard_conforming_strings = off;

diff --git a/src/backend/optimizer/util/clauses.c b/src/backend/optimizer/util/clauses.c
index 2d3ec22407..5ce0079a12 100644
--- a/src/backend/optimizer/util/clauses.c
+++ b/src/backend/optimizer/util/clauses.c
@@ -108,6 +108,7 @@ static bool contain_volatile_functions_not_nextval_walker(Node *node, void *cont
 static bool max_parallel_hazard_walker(Node *node,
 									   max_parallel_hazard_context *context);
 static bool contain_nonstrict_functions_walker(Node *node, void *context);
+static bool contain_inlinable_functions_walker(Node *node, void *context);
 static bool contain_context_dependent_node(Node *clause);
 static bool contain_context_dependent_node_walker(Node *node, int *flags);
 static bool contain_leaked_vars_walker(Node *node, void *context);
@@ -1218,6 +1219,63 @@ contain_nonstrict_functions_walker(Node *node, void *context)
 								  context);
 }
 
+/*****************************************************************************
+ *		Check clauses for inline-able functions
+ *****************************************************************************/
+
+bool
+contain_inlinable_functions(Node *node)
+{
+	return contain_inlinable_functions_walker(node, NULL);
+}
+
+/*
+ * can_inline_function - checks if a function is inline-able
+ */
+static bool
+can_inline_function(HeapTuple func_tuple)
+{
+	Form_pg_proc funcform = (Form_pg_proc) GETSTRUCT(func_tuple);
+
+	/*
+	 * Nope if the function is not SQL-language or has other showstopper
+	 * properties.  (The prokind and nargs checks are just paranoia.)
+	 */
+	return	funcform->prolang == SQLlanguageId &&
+			funcform->prokind == PROKIND_FUNCTION &&
+			!funcform->prosecdef && !funcform->proretset &&
+			funcform->prorettype != RECORDOID &&
+			heap_attisnull(func_tuple, Anum_pg_proc_proconfig, NULL);
+}
+
+static bool
+can_inline_function_checker(Oid funcid, void *context)
+{
+	HeapTuple	func_tuple;
+	bool		result;
+
+	func_tuple = SearchSysCache1(PROCOID, ObjectIdGetDatum(funcid));
+	if (!HeapTupleIsValid(func_tuple))
+		elog(ERROR, "cache lookup failed for function %u", funcid);
+
+	result = can_inline_function(func_tuple);
+	ReleaseSysCache(func_tuple);
+
+	return result;
+}
+
+static bool
+contain_inlinable_functions_walker(Node *node, void *context)
+{
+	if (node == NULL)
+		return false;
+	if (check_functions_in_node(node, can_inline_function_checker, context))
+		return true;
+
+	return expression_tree_walker(node, contain_inlinable_functions_walker,
+								  context);
+}
+
 /*****************************************************************************
  *		Check clauses for context-dependent nodes
  *****************************************************************************/
@@ -4022,7 +4080,8 @@ simplify_function(Oid funcid, Oid result_type, int32 result_typmod,
 		Assert(newexpr != (Expr *) &fexpr);
 	}
 
-	if (!newexpr && allow_non_const)
+	if (!newexpr && allow_non_const &&
+		can_inline_function(func_tuple))
 		newexpr = inline_function(funcid, result_type, result_collid,
 								  input_collid, args, funcvariadic,
 								  func_tuple, context);
@@ -4415,16 +4474,11 @@ inline_function(Oid funcid, Oid result_type, Oid result_collid,
 	int			i;
 
 	/*
-	 * Forget it if the function is not SQL-language or has other showstopper
-	 * properties.  (The prokind and nargs checks are just paranoia.)
+	 * Caller should already have checked whether the function can be inlined
+	 * using can_function_inline().
 	 */
-	if (funcform->prolang != SQLlanguageId ||
-		funcform->prokind != PROKIND_FUNCTION ||
-		funcform->prosecdef ||
-		funcform->proretset ||
-		funcform->prorettype == RECORDOID ||
-		!heap_attisnull(func_tuple, Anum_pg_proc_proconfig, NULL) ||
-		funcform->pronargs != list_length(args))
+
+	if (funcform->pronargs != list_length(args))
 		return NULL;
 
 	/* Check for recursive function, and give up trying to expand if so */
diff --git a/src/include/optimizer/optimizer.h b/src/include/optimizer/optimizer.h
index 5283995df8..4954657147 100644
--- a/src/include/optimizer/optimizer.h
+++ b/src/include/optimizer/optimizer.h
@@ -130,6 +130,7 @@ extern Expr *canonicalize_qual(Expr *qual, bool is_check);
 extern bool contain_mutable_functions(Node *clause);
 extern bool contain_volatile_functions(Node *clause);
 extern bool contain_volatile_functions_not_nextval(Node *clause);
+extern bool contain_inlinable_functions(Node *node);
 
 extern Node *eval_const_expressions(PlannerInfo *root, Node *node);
 
diff --git a/src/pl/plpgsql/src/pl_exec.c b/src/pl/plpgsql/src/pl_exec.c
index 5acf604f63..07d468ce5d 100644
--- a/src/pl/plpgsql/src/pl_exec.c
+++ b/src/pl/plpgsql/src/pl_exec.c
@@ -5758,9 +5758,10 @@ exec_eval_expr(PLpgSQL_execstate *estate,
 	Form_pg_attribute attr;
 
 	/*
-	 * If first time through, create a plan for this expression.
+	 * Create a plan for this expression, if first time through or the
+	 * existing plan is no longer valid.
 	 */
-	if (expr->plan == NULL)
+	if (expr->plan == NULL || !SPI_plan_is_valid(expr->plan))
 		exec_prepare_plan(estate, expr, CURSOR_OPT_PARALLEL_OK, true);
 
 	/*
@@ -6076,7 +6077,6 @@ exec_eval_simple_expr(PLpgSQL_execstate *estate,
 {
 	ExprContext *econtext = estate->eval_econtext;
 	LocalTransactionId curlxid = MyProc->lxid;
-	CachedPlan *cplan;
 	void	   *save_setup_arg;
 	bool		need_snapshot;
 	MemoryContext oldcontext;
@@ -6094,28 +6094,39 @@ exec_eval_simple_expr(PLpgSQL_execstate *estate,
 		return false;
 
 	/*
-	 * Revalidate cached plan, so that we will notice if it became stale. (We
-	 * need to hold a refcount while using the plan, anyway.)  If replanning
-	 * is needed, do that work in the eval_mcontext.
+	 * Revalidate cached plan if one would have been used (due to inline-able)
+	 * functions being found in the expression), so that we will notice if it
+	 * became stale. (We need to hold a refcount while using the plan,
+	 * anyway.)  If replanning is needed, do that work in the eval_mcontext.
 	 */
-	oldcontext = MemoryContextSwitchTo(get_eval_mcontext(estate));
-	cplan = SPI_plan_get_cached_plan(expr->plan);
-	MemoryContextSwitchTo(oldcontext);
-
-	/*
-	 * We can't get a failure here, because the number of CachedPlanSources in
-	 * the SPI plan can't change from what exec_simple_check_plan saw; it's a
-	 * property of the raw parsetree generated from the query text.
-	 */
-	Assert(cplan != NULL);
-
-	/* If it got replanned, update our copy of the simple expression */
-	if (cplan->generation != expr->expr_simple_generation)
+	if (expr->expr_contains_inline_func)
 	{
-		exec_save_simple_expr(expr, cplan);
-		/* better recheck r/w safety, as it could change due to inlining */
-		if (expr->rwparam >= 0)
-			exec_check_rw_parameter(expr, expr->rwparam);
+		CachedPlan *cplan;
+
+		oldcontext = MemoryContextSwitchTo(get_eval_mcontext(estate));
+		cplan = SPI_plan_get_cached_plan(expr->plan);
+		MemoryContextSwitchTo(oldcontext);
+
+		/*
+		 * We can't get a failure here, because the number of CachedPlanSources in
+		 * the SPI plan can't change from what exec_simple_check_plan saw; it's a
+		 * property of the raw parsetree generated from the query text.
+		 */
+		Assert(cplan != NULL);
+
+		/* If it got replanned, update our copy of the simple expression */
+		if (cplan->generation != expr->expr_simple_generation)
+		{
+			exec_save_simple_expr(expr, cplan);
+			/* better recheck r/w safety, as it could change due to inlining */
+			if (expr->rwparam >= 0)
+				exec_check_rw_parameter(expr, expr->rwparam);
+		}
+
+		/*
+		 * Now we can release our refcount on the cached plan.
+		 */
+		ReleaseCachedPlan(cplan, true);
 	}
 
 	/*
@@ -6192,11 +6203,6 @@ exec_eval_simple_expr(PLpgSQL_execstate *estate,
 
 	MemoryContextSwitchTo(oldcontext);
 
-	/*
-	 * Now we can release our refcount on the cached plan.
-	 */
-	ReleaseCachedPlan(cplan, true);
-
 	/*
 	 * That's it.
 	 */
@@ -7893,6 +7899,8 @@ get_cast_hashentry(PLpgSQL_execstate *estate,
  * exec_simple_check_plan -		Check if a plan is simple enough to
  *								be evaluated by ExecEvalExpr() instead
  *								of SPI.
+ *
+ * If it is, set expr->expr_simple_expr.
  * ----------
  */
 static void
@@ -7901,8 +7909,7 @@ exec_simple_check_plan(PLpgSQL_execstate *estate, PLpgSQL_expr *expr)
 	List	   *plansources;
 	CachedPlanSource *plansource;
 	Query	   *query;
-	CachedPlan *cplan;
-	MemoryContext oldcontext;
+	Expr	   *tle_expr;
 
 	/*
 	 * Initialize to "not simple".
@@ -7972,23 +7979,55 @@ exec_simple_check_plan(PLpgSQL_execstate *estate, PLpgSQL_expr *expr)
 		return;
 
 	/*
-	 * OK, we can treat it as a simple plan.
-	 *
-	 * Get the generic plan for the query.  If replanning is needed, do that
-	 * work in the eval_mcontext.
+	 * We have a simple expression.  Although, if it contains inline-able
+	 * SQL functions, better pass it through the planner to get a simpler
+	 * plan.
 	 */
-	oldcontext = MemoryContextSwitchTo(get_eval_mcontext(estate));
-	cplan = SPI_plan_get_cached_plan(expr->plan);
-	MemoryContextSwitchTo(oldcontext);
+	tle_expr = castNode(TargetEntry, linitial(query->targetList))->expr;
+	if (contain_inlinable_functions((Node *) tle_expr))
+	{
+		CachedPlan *cplan = NULL;
+		MemoryContext oldcontext;
 
-	/* Can't fail, because we checked for a single CachedPlanSource above */
-	Assert(cplan != NULL);
+		expr->expr_contains_inline_func = true;
 
-	/* Share the remaining work with replan code path */
-	exec_save_simple_expr(expr, cplan);
+		/*
+		 * Get the generic plan for the query.  If replanning is needed, do
+		 * that work in the eval_mcontext.
+		 */
+		oldcontext = MemoryContextSwitchTo(get_eval_mcontext(estate));
+		cplan = SPI_plan_get_cached_plan(expr->plan);
+		MemoryContextSwitchTo(oldcontext);
 
-	/* Release our plan refcount */
-	ReleaseCachedPlan(cplan, true);
+		/*
+		 * Can't fail, because we checked for a single CachedPlanSource
+		 * above.
+		 */
+		Assert(cplan != NULL);
+
+		exec_save_simple_expr(expr, cplan);
+
+		/* Release our plan refcount */
+		ReleaseCachedPlan(cplan, true);
+	}
+	else
+	{
+		/*
+		 * Save the simple expression, and initialize state to "not valid in
+		 * current transaction".
+		 */
+		expr->expr_simple_expr = tle_expr;
+		expr->expr_simple_generation = plansource->generation;
+		expr->expr_simple_state = NULL;
+		expr->expr_simple_in_use = false;
+		expr->expr_simple_lxid = InvalidLocalTransactionId;
+		/* Also stash away the expression result type */
+		expr->expr_simple_type = exprType((Node *) tle_expr);
+		expr->expr_simple_typmod = exprTypmod((Node *) tle_expr);
+		/* We also want to remember if it is immutable or not */
+		expr->expr_simple_mutable = contain_mutable_functions((Node *) tle_expr);
+		expr->expr_contains_inline_func = false;
+	}
 }
 
 /*
diff --git a/src/pl/plpgsql/src/plpgsql.h b/src/pl/plpgsql/src/plpgsql.h
index b599f67fc5..0e02212135 100644
--- a/src/pl/plpgsql/src/plpgsql.h
+++ b/src/pl/plpgsql/src/plpgsql.h
@@ -245,6 +245,7 @@ typedef struct PLpgSQL_expr
 	ExprState  *expr_simple_state;	/* eval tree for expr_simple_expr */
 	bool		expr_simple_in_use; /* true if eval tree is active */
 	LocalTransactionId expr_simple_lxid;
+	bool		expr_contains_inline_func;
 } PLpgSQL_expr;
 
 /*
diff --git a/src/test/regress/expected/plpgsql.out b/src/test/regress/expected/plpgsql.out
index cd2c79f4d5..3abbc7bc57 100644
--- a/src/test/regress/expected/plpgsql.out
+++ b/src/test/regress/expected/plpgsql.out
@@ -4289,12 +4289,10 @@ end
 $$;
 select fail();
 ERROR:  division by zero
-CONTEXT:  SQL statement "SELECT 1/0"
-PL/pgSQL function fail() line 3 at RETURN
+CONTEXT:  PL/pgSQL function fail() line 3 at RETURN
 select fail();
 ERROR:  division by zero
-CONTEXT:  SQL statement "SELECT 1/0"
-PL/pgSQL function fail() line 3 at RETURN
+CONTEXT:  PL/pgSQL function fail() line 3 at RETURN
 drop function fail();
 -- Test handling of string literals.
 set standard_conforming_strings = off;

diff -U3 /home/pavel/src/postgresql.master/src/pl/plpgsql/src/expected/plpgsql_varprops.out /home/pavel/src/postgresql.master/src/pl/plpgsql/src/results/plpgsql_varprops.out
--- /home/pavel/src/postgresql.master/src/pl/plpgsql/src/expected/plpgsql_varprops.out	2020-02-16 16:27:36.879112285 +0100
+++ /home/pavel/src/postgresql.master/src/pl/plpgsql/src/results/plpgsql_varprops.out	2020-02-19 07:34:07.134558929 +0100
@@ -76,8 +76,7 @@
   raise notice 'x = %', x;
 end$$;
 ERROR:  division by zero
-CONTEXT:  SQL statement "SELECT 1/0"
-PL/pgSQL function inline_code_block line 3 during statement block local variable initialization
+CONTEXT:  PL/pgSQL function inline_code_block line 3 during statement block local variable initialization
 do $$
 declare x bigint[] := array[1,3,5];
 begin

diff --git a/src/backend/optimizer/util/clauses.c b/src/backend/optimizer/util/clauses.c
index 2d3ec22407..5ce0079a12 100644
--- a/src/backend/optimizer/util/clauses.c
+++ b/src/backend/optimizer/util/clauses.c
@@ -108,6 +108,7 @@ static bool contain_volatile_functions_not_nextval_walker(Node *node, void *cont
 static bool max_parallel_hazard_walker(Node *node,
 									   max_parallel_hazard_context *context);
 static bool contain_nonstrict_functions_walker(Node *node, void *context);
+static bool contain_inlinable_functions_walker(Node *node, void *context);
 static bool contain_context_dependent_node(Node *clause);
 static bool contain_context_dependent_node_walker(Node *node, int *flags);
 static bool contain_leaked_vars_walker(Node *node, void *context);
@@ -1218,6 +1219,63 @@ contain_nonstrict_functions_walker(Node *node, void *context)
 								  context);
 }
 
+/*****************************************************************************
+ *		Check clauses for inline-able functions
+ *****************************************************************************/
+
+bool
+contain_inlinable_functions(Node *node)
+{
+	return contain_inlinable_functions_walker(node, NULL);
+}
+
+/*
+ * can_inline_function - checks if a function is inline-able
+ */
+static bool
+can_inline_function(HeapTuple func_tuple)
+{
+	Form_pg_proc funcform = (Form_pg_proc) GETSTRUCT(func_tuple);
+
+	/*
+	 * Nope if the function is not SQL-language or has other showstopper
+	 * properties.  (The prokind and nargs checks are just paranoia.)
+	 */
+	return	funcform->prolang == SQLlanguageId &&
+			funcform->prokind == PROKIND_FUNCTION &&
+			!funcform->prosecdef && !funcform->proretset &&
+			funcform->prorettype != RECORDOID &&
+			heap_attisnull(func_tuple, Anum_pg_proc_proconfig, NULL);
+}
+
+static bool
+can_inline_function_checker(Oid funcid, void *context)
+{
+	HeapTuple	func_tuple;
+	bool		result;
+
+	func_tuple = SearchSysCache1(PROCOID, ObjectIdGetDatum(funcid));
+	if (!HeapTupleIsValid(func_tuple))
+		elog(ERROR, "cache lookup failed for function %u", funcid);
+
+	result = can_inline_function(func_tuple);
+	ReleaseSysCache(func_tuple);
+
+	return result;
+}
+
+static bool
+contain_inlinable_functions_walker(Node *node, void *context)
+{
+	if (node == NULL)
+		return false;
+	if (check_functions_in_node(node, can_inline_function_checker, context))
+		return true;
+
+	return expression_tree_walker(node, contain_inlinable_functions_walker,
+								  context);
+}
+
 /*****************************************************************************
  *		Check clauses for context-dependent nodes
  *****************************************************************************/
@@ -4022,7 +4080,8 @@ simplify_function(Oid funcid, Oid result_type, int32 result_typmod,
 		Assert(newexpr != (Expr *) &fexpr);
 	}
 
-	if (!newexpr && allow_non_const)
+	if (!newexpr && allow_non_const &&
+		can_inline_function(func_tuple))
 		newexpr = inline_function(funcid, result_type, result_collid,
 								  input_collid, args, funcvariadic,
 								  func_tuple, context);
@@ -4415,16 +4474,11 @@ inline_function(Oid funcid, Oid result_type, Oid result_collid,
 	int			i;
 
 	/*
-	 * Forget it if the function is not SQL-language or has other showstopper
-	 * properties.  (The prokind and nargs checks are just paranoia.)
+	 * Caller should already have checked whether the function can be inlined
+	 * using can_function_inline().
 	 */
-	if (funcform->prolang != SQLlanguageId ||
-		funcform->prokind != PROKIND_FUNCTION ||
-		funcform->prosecdef ||
-		funcform->proretset ||
-		funcform->prorettype == RECORDOID ||
-		!heap_attisnull(func_tuple, Anum_pg_proc_proconfig, NULL) ||
-		funcform->pronargs != list_length(args))
+
+	if (funcform->pronargs != list_length(args))
 		return NULL;
 
 	/* Check for recursive function, and give up trying to expand if so */
diff --git a/src/include/optimizer/optimizer.h b/src/include/optimizer/optimizer.h
index 5283995df8..4954657147 100644
--- a/src/include/optimizer/optimizer.h
+++ b/src/include/optimizer/optimizer.h
@@ -130,6 +130,7 @@ extern Expr *canonicalize_qual(Expr *qual, bool is_check);
 extern bool contain_mutable_functions(Node *clause);
 extern bool contain_volatile_functions(Node *clause);
 extern bool contain_volatile_functions_not_nextval(Node *clause);
+extern bool contain_inlinable_functions(Node *node);
 
 extern Node *eval_const_expressions(PlannerInfo *root, Node *node);
 
diff --git a/src/pl/plpgsql/src/expected/plpgsql_varprops.out b/src/pl/plpgsql/src/expected/plpgsql_varprops.out
index 18f03d75b4..84a4ea7d03 100644
--- a/src/pl/plpgsql/src/expected/plpgsql_varprops.out
+++ b/src/pl/plpgsql/src/expected/plpgsql_varprops.out
@@ -76,8 +76,7 @@ begin
   raise notice 'x = %', x;
 end$$;
 ERROR:  division by zero
-CONTEXT:  SQL statement "SELECT 1/0"
-PL/pgSQL function inline_code_block line 3 during statement block local variable initialization
+CONTEXT:  PL/pgSQL function inline_code_block line 3 during statement block local variable initialization
 do $$
 declare x bigint[] := array[1,3,5];
 begin
diff --git a/src/pl/plpgsql/src/pl_exec.c b/src/pl/plpgsql/src/pl_exec.c
index 5acf604f63..07d468ce5d 100644
--- a/src/pl/plpgsql/src/pl_exec.c
+++ b/src/pl/plpgsql/src/pl_exec.c
@@ -5758,9 +5758,10 @@ exec_eval_expr(PLpgSQL_execstate *estate,
 	Form_pg_attribute attr;
 
 	/*
-	 * If first time through, create a plan for this expression.
+	 * Create a plan for this expression, if first time through or the
+	 * existing plan is no longer valid.
 	 */
-	if (expr->plan == NULL)
+	if (expr->plan == NULL || !SPI_plan_is_valid(expr->plan))
 		exec_prepare_plan(estate, expr, CURSOR_OPT_PARALLEL_OK, true);
 
 	/*
@@ -6076,7 +6077,6 @@ exec_eval_simple_expr(PLpgSQL_execstate *estate,
 {
 	ExprContext *econtext = estate->eval_econtext;
 	LocalTransactionId curlxid = MyProc->lxid;
-	CachedPlan *cplan;
 	void	   *save_setup_arg;
 	bool		need_snapshot;
 	MemoryContext oldcontext;
@@ -6094,28 +6094,39 @@ exec_eval_simple_expr(PLpgSQL_execstate *estate,
 		return false;
 
 	/*
-	 * Revalidate cached plan, so that we will notice if it became stale. (We
-	 * need to hold a refcount while using the plan, anyway.)  If replanning
-	 * is needed, do that work in the eval_mcontext.
+	 * Revalidate cached plan if one would have been used (due to inline-able)
+	 * functions being found in the expression), so that we will notice if it
+	 * became stale. (We need to hold a refcount while using the plan,
+	 * anyway.)  If replanning is needed, do that work in the eval_mcontext.
 	 */
-	oldcontext = MemoryContextSwitchTo(get_eval_mcontext(estate));
-	cplan = SPI_plan_get_cached_plan(expr->plan);
-	MemoryContextSwitchTo(oldcontext);
+	if (expr->expr_contains_inline_func)
+	{
+		CachedPlan *cplan;
 
-	/*
-	 * We can't get a failure here, because the number of CachedPlanSources in
-	 * the SPI plan can't change from what exec_simple_check_plan saw; it's a
-	 * property of the raw parsetree generated from the query text.
-	 */
-	Assert(cplan != NULL);
+		oldcontext = MemoryContextSwitchTo(get_eval_mcontext(estate));
+		cplan = SPI_plan_get_cached_plan(expr->plan);
+		MemoryContextSwitchTo(oldcontext);
 
-	/* If it got replanned, update our copy of the simple expression */
-	if (cplan->generation != expr->expr_simple_generation)
-	{
-		exec_save_simple_expr(expr, cplan);
-		/* better recheck r/w safety, as it could change due to inlining */
-		if (expr->rwparam >= 0)
-			exec_check_rw_parameter(expr, expr->rwparam);
+		/*
+		 * We can't get a failure here, because the number of CachedPlanSources in
+		 * the SPI plan can't change from what exec_simple_check_plan saw; it's a
+		 * property of the raw parsetree generated from the query text.
+		 */
+		Assert(cplan != NULL);
+
+		/* If it got replanned, update our copy of the simple expression */
+		if (cplan->generation != expr->expr_simple_generation)
+		{
+			exec_save_simple_expr(expr, cplan);
+			/* better recheck r/w safety, as it could change due to inlining */
+			if (expr->rwparam >= 0)
+				exec_check_rw_parameter(expr, expr->rwparam);
+		}
+
+		/*
+		 * Now we can release our refcount on the cached plan.
+		 */
+		ReleaseCachedPlan(cplan, true);
 	}
 
 	/*
@@ -6192,11 +6203,6 @@ exec_eval_simple_expr(PLpgSQL_execstate *estate,
 
 	MemoryContextSwitchTo(oldcontext);
 
-	/*
-	 * Now we can release our refcount on the cached plan.
-	 */
-	ReleaseCachedPlan(cplan, true);
-
 	/*
 	 * That's it.
 	 */
@@ -7893,6 +7899,8 @@ get_cast_hashentry(PLpgSQL_execstate *estate,
  * exec_simple_check_plan -		Check if a plan is simple enough to
  *								be evaluated by ExecEvalExpr() instead
  *								of SPI.
+ *
+ * If it is, set expr->expr_simple_expr.
  * ----------
  */
 static void
@@ -7901,8 +7909,7 @@ exec_simple_check_plan(PLpgSQL_execstate *estate, PLpgSQL_expr *expr)
 	List	   *plansources;
 	CachedPlanSource *plansource;
 	Query	   *query;
-	CachedPlan *cplan;
-	MemoryContext oldcontext;
+	Expr	   *tle_expr;
 
 	/*
 	 * Initialize to "not simple".
@@ -7972,23 +7979,55 @@ exec_simple_check_plan(PLpgSQL_execstate *estate, PLpgSQL_expr *expr)
 		return;
 
 	/*
-	 * OK, we can treat it as a simple plan.
-	 *
-	 * Get the generic plan for the query.  If replanning is needed, do that
-	 * work in the eval_mcontext.
+	 * We have a simple expression.  Although, if it contains inline-able
+	 * SQL functions, better pass it through the planner to get a simpler
+	 * plan.
 	 */
-	oldcontext = MemoryContextSwitchTo(get_eval_mcontext(estate));
-	cplan = SPI_plan_get_cached_plan(expr->plan);
-	MemoryContextSwitchTo(oldcontext);
+	tle_expr = castNode(TargetEntry, linitial(query->targetList))->expr;
+	if (contain_inlinable_functions((Node *) tle_expr))
+	{
+		CachedPlan *cplan = NULL;
+		MemoryContext oldcontext;
+
+		expr->expr_contains_inline_func = true;
+
+		/*
+		 * Get the generic plan for the query.  If replanning is needed, do
+		 * that work in the eval_mcontext.
+		 */
+		oldcontext = MemoryContextSwitchTo(get_eval_mcontext(estate));
+		cplan = SPI_plan_get_cached_plan(expr->plan);
+		MemoryContextSwitchTo(oldcontext);
 
-	/* Can't fail, because we checked for a single CachedPlanSource above */
-	Assert(cplan != NULL);
+		/*
+		 * Can't fail, because we checked for a single CachedPlanSource
+		 * above.
+		 */
+		Assert(cplan != NULL);
 
-	/* Share the remaining work with replan code path */
-	exec_save_simple_expr(expr, cplan);
+		exec_save_simple_expr(expr, cplan);
 
-	/* Release our plan refcount */
-	ReleaseCachedPlan(cplan, true);
+		/* Release our plan refcount */
+		ReleaseCachedPlan(cplan, true);
+	}
+	else
+	{
+		/*
+		 * Save the simple expression, and initialize state to "not valid in
+		 * current transaction".
+		 */
+		expr->expr_simple_expr = tle_expr;
+		expr->expr_simple_generation = plansource->generation;
+		expr->expr_simple_state = NULL;
+		expr->expr_simple_in_use = false;
+		expr->expr_simple_lxid = InvalidLocalTransactionId;
+		/* Also stash away the expression result type */
+		expr->expr_simple_type = exprType((Node *) tle_expr);
+		expr->expr_simple_typmod = exprTypmod((Node *) tle_expr);
+		/* We also want to remember if it is immutable or not */
+		expr->expr_simple_mutable = contain_mutable_functions((Node *) tle_expr);
+		expr->expr_contains_inline_func = false;
+	}
 }
 
 /*
diff --git a/src/pl/plpgsql/src/plpgsql.h b/src/pl/plpgsql/src/plpgsql.h
index b599f67fc5..0e02212135 100644
--- a/src/pl/plpgsql/src/plpgsql.h
+++ b/src/pl/plpgsql/src/plpgsql.h
@@ -245,6 +245,7 @@ typedef struct PLpgSQL_expr
 	ExprState  *expr_simple_state;	/* eval tree for expr_simple_expr */
 	bool		expr_simple_in_use; /* true if eval tree is active */
 	LocalTransactionId expr_simple_lxid;
+	bool		expr_contains_inline_func;
 } PLpgSQL_expr;
 
 /*
diff --git a/src/test/regress/expected/plpgsql.out b/src/test/regress/expected/plpgsql.out
index cd2c79f4d5..3abbc7bc57 100644
--- a/src/test/regress/expected/plpgsql.out
+++ b/src/test/regress/expected/plpgsql.out
@@ -4289,12 +4289,10 @@ end
 $$;
 select fail();
 ERROR:  division by zero
-CONTEXT:  SQL statement "SELECT 1/0"
-PL/pgSQL function fail() line 3 at RETURN
+CONTEXT:  PL/pgSQL function fail() line 3 at RETURN
 select fail();
 ERROR:  division by zero
-CONTEXT:  SQL statement "SELECT 1/0"
-PL/pgSQL function fail() line 3 at RETURN
+CONTEXT:  PL/pgSQL function fail() line 3 at RETURN
 drop function fail();
 -- Test handling of string literals.
 set standard_conforming_strings = off;

diff --git a/src/backend/optimizer/util/clauses.c b/src/backend/optimizer/util/clauses.c
index 2d3ec22407..5ce0079a12 100644
--- a/src/backend/optimizer/util/clauses.c
+++ b/src/backend/optimizer/util/clauses.c
@@ -108,6 +108,7 @@ static bool contain_volatile_functions_not_nextval_walker(Node *node, void *cont
 static bool max_parallel_hazard_walker(Node *node,
 									   max_parallel_hazard_context *context);
 static bool contain_nonstrict_functions_walker(Node *node, void *context);
+static bool contain_inlinable_functions_walker(Node *node, void *context);
 static bool contain_context_dependent_node(Node *clause);
 static bool contain_context_dependent_node_walker(Node *node, int *flags);
 static bool contain_leaked_vars_walker(Node *node, void *context);
@@ -1218,6 +1219,63 @@ contain_nonstrict_functions_walker(Node *node, void *context)
 								  context);
 }
 
+/*****************************************************************************
+ *		Check clauses for inline-able functions
+ *****************************************************************************/
+
+bool
+contain_inlinable_functions(Node *node)
+{
+	return contain_inlinable_functions_walker(node, NULL);
+}
+
+/*
+ * can_inline_function - checks if a function is inline-able
+ */
+static bool
+can_inline_function(HeapTuple func_tuple)
+{
+	Form_pg_proc funcform = (Form_pg_proc) GETSTRUCT(func_tuple);
+
+	/*
+	 * Nope if the function is not SQL-language or has other showstopper
+	 * properties.  (The prokind and nargs checks are just paranoia.)
+	 */
+	return	funcform->prolang == SQLlanguageId &&
+			funcform->prokind == PROKIND_FUNCTION &&
+			!funcform->prosecdef && !funcform->proretset &&
+			funcform->prorettype != RECORDOID &&
+			heap_attisnull(func_tuple, Anum_pg_proc_proconfig, NULL);
+}
+
+static bool
+can_inline_function_checker(Oid funcid, void *context)
+{
+	HeapTuple	func_tuple;
+	bool		result;
+
+	func_tuple = SearchSysCache1(PROCOID, ObjectIdGetDatum(funcid));
+	if (!HeapTupleIsValid(func_tuple))
+		elog(ERROR, "cache lookup failed for function %u", funcid);
+
+	result = can_inline_function(func_tuple);
+	ReleaseSysCache(func_tuple);
+
+	return result;
+}
+
+static bool
+contain_inlinable_functions_walker(Node *node, void *context)
+{
+	if (node == NULL)
+		return false;
+	if (check_functions_in_node(node, can_inline_function_checker, context))
+		return true;
+
+	return expression_tree_walker(node, contain_inlinable_functions_walker,
+								  context);
+}
+
 /*****************************************************************************
  *		Check clauses for context-dependent nodes
  *****************************************************************************/
@@ -4022,7 +4080,8 @@ simplify_function(Oid funcid, Oid result_type, int32 result_typmod,
 		Assert(newexpr != (Expr *) &fexpr);
 	}
 
-	if (!newexpr && allow_non_const)
+	if (!newexpr && allow_non_const &&
+		can_inline_function(func_tuple))
 		newexpr = inline_function(funcid, result_type, result_collid,
 								  input_collid, args, funcvariadic,
 								  func_tuple, context);
@@ -4415,16 +4474,11 @@ inline_function(Oid funcid, Oid result_type, Oid result_collid,
 	int			i;
 
 	/*
-	 * Forget it if the function is not SQL-language or has other showstopper
-	 * properties.  (The prokind and nargs checks are just paranoia.)
+	 * Caller should already have checked whether the function can be inlined
+	 * using can_function_inline().
 	 */
-	if (funcform->prolang != SQLlanguageId ||
-		funcform->prokind != PROKIND_FUNCTION ||
-		funcform->prosecdef ||
-		funcform->proretset ||
-		funcform->prorettype == RECORDOID ||
-		!heap_attisnull(func_tuple, Anum_pg_proc_proconfig, NULL) ||
-		funcform->pronargs != list_length(args))
+
+	if (funcform->pronargs != list_length(args))
 		return NULL;
 
 	/* Check for recursive function, and give up trying to expand if so */
diff --git a/src/include/optimizer/optimizer.h b/src/include/optimizer/optimizer.h
index 5283995df8..4954657147 100644
--- a/src/include/optimizer/optimizer.h
+++ b/src/include/optimizer/optimizer.h
@@ -130,6 +130,7 @@ extern Expr *canonicalize_qual(Expr *qual, bool is_check);
 extern bool contain_mutable_functions(Node *clause);
 extern bool contain_volatile_functions(Node *clause);
 extern bool contain_volatile_functions_not_nextval(Node *clause);
+extern bool contain_inlinable_functions(Node *node);
 
 extern Node *eval_const_expressions(PlannerInfo *root, Node *node);
 
diff --git a/src/pl/plpgsql/src/expected/plpgsql_varprops.out b/src/pl/plpgsql/src/expected/plpgsql_varprops.out
index 18f03d75b4..84a4ea7d03 100644
--- a/src/pl/plpgsql/src/expected/plpgsql_varprops.out
+++ b/src/pl/plpgsql/src/expected/plpgsql_varprops.out
@@ -76,8 +76,7 @@ begin
   raise notice 'x = %', x;
 end$$;
 ERROR:  division by zero
-CONTEXT:  SQL statement "SELECT 1/0"
-PL/pgSQL function inline_code_block line 3 during statement block local variable initialization
+CONTEXT:  PL/pgSQL function inline_code_block line 3 during statement block local variable initialization
 do $$
 declare x bigint[] := array[1,3,5];
 begin
diff --git a/src/pl/plpgsql/src/pl_exec.c b/src/pl/plpgsql/src/pl_exec.c
index 5acf604f63..07d468ce5d 100644
--- a/src/pl/plpgsql/src/pl_exec.c
+++ b/src/pl/plpgsql/src/pl_exec.c
@@ -5758,9 +5758,10 @@ exec_eval_expr(PLpgSQL_execstate *estate,
 	Form_pg_attribute attr;
 
 	/*
-	 * If first time through, create a plan for this expression.
+	 * Create a plan for this expression, if first time through or the
+	 * existing plan is no longer valid.
 	 */
-	if (expr->plan == NULL)
+	if (expr->plan == NULL || !SPI_plan_is_valid(expr->plan))
 		exec_prepare_plan(estate, expr, CURSOR_OPT_PARALLEL_OK, true);
 
 	/*
@@ -6076,7 +6077,6 @@ exec_eval_simple_expr(PLpgSQL_execstate *estate,
 {
 	ExprContext *econtext = estate->eval_econtext;
 	LocalTransactionId curlxid = MyProc->lxid;
-	CachedPlan *cplan;
 	void	   *save_setup_arg;
 	bool		need_snapshot;
 	MemoryContext oldcontext;
@@ -6094,28 +6094,39 @@ exec_eval_simple_expr(PLpgSQL_execstate *estate,
 		return false;
 
 	/*
-	 * Revalidate cached plan, so that we will notice if it became stale. (We
-	 * need to hold a refcount while using the plan, anyway.)  If replanning
-	 * is needed, do that work in the eval_mcontext.
+	 * Revalidate cached plan if one would have been used (due to inline-able)
+	 * functions being found in the expression), so that we will notice if it
+	 * became stale. (We need to hold a refcount while using the plan,
+	 * anyway.)  If replanning is needed, do that work in the eval_mcontext.
 	 */
-	oldcontext = MemoryContextSwitchTo(get_eval_mcontext(estate));
-	cplan = SPI_plan_get_cached_plan(expr->plan);
-	MemoryContextSwitchTo(oldcontext);
+	if (expr->expr_contains_inline_func)
+	{
+		CachedPlan *cplan;
 
-	/*
-	 * We can't get a failure here, because the number of CachedPlanSources in
-	 * the SPI plan can't change from what exec_simple_check_plan saw; it's a
-	 * property of the raw parsetree generated from the query text.
-	 */
-	Assert(cplan != NULL);
+		oldcontext = MemoryContextSwitchTo(get_eval_mcontext(estate));
+		cplan = SPI_plan_get_cached_plan(expr->plan);
+		MemoryContextSwitchTo(oldcontext);
 
-	/* If it got replanned, update our copy of the simple expression */
-	if (cplan->generation != expr->expr_simple_generation)
-	{
-		exec_save_simple_expr(expr, cplan);
-		/* better recheck r/w safety, as it could change due to inlining */
-		if (expr->rwparam >= 0)
-			exec_check_rw_parameter(expr, expr->rwparam);
+		/*
+		 * We can't get a failure here, because the number of CachedPlanSources in
+		 * the SPI plan can't change from what exec_simple_check_plan saw; it's a
+		 * property of the raw parsetree generated from the query text.
+		 */
+		Assert(cplan != NULL);
+
+		/* If it got replanned, update our copy of the simple expression */
+		if (cplan->generation != expr->expr_simple_generation)
+		{
+			exec_save_simple_expr(expr, cplan);
+			/* better recheck r/w safety, as it could change due to inlining */
+			if (expr->rwparam >= 0)
+				exec_check_rw_parameter(expr, expr->rwparam);
+		}
+
+		/*
+		 * Now we can release our refcount on the cached plan.
+		 */
+		ReleaseCachedPlan(cplan, true);
 	}
 
 	/*
@@ -6192,11 +6203,6 @@ exec_eval_simple_expr(PLpgSQL_execstate *estate,
 
 	MemoryContextSwitchTo(oldcontext);
 
-	/*
-	 * Now we can release our refcount on the cached plan.
-	 */
-	ReleaseCachedPlan(cplan, true);
-
 	/*
 	 * That's it.
 	 */
@@ -7893,6 +7899,8 @@ get_cast_hashentry(PLpgSQL_execstate *estate,
  * exec_simple_check_plan -		Check if a plan is simple enough to
  *								be evaluated by ExecEvalExpr() instead
  *								of SPI.
+ *
+ * If it is, set expr->expr_simple_expr.
  * ----------
  */
 static void
@@ -7901,8 +7909,7 @@ exec_simple_check_plan(PLpgSQL_execstate *estate, PLpgSQL_expr *expr)
 	List	   *plansources;
 	CachedPlanSource *plansource;
 	Query	   *query;
-	CachedPlan *cplan;
-	MemoryContext oldcontext;
+	Expr	   *tle_expr;
 
 	/*
 	 * Initialize to "not simple".
@@ -7972,23 +7979,55 @@ exec_simple_check_plan(PLpgSQL_execstate *estate, PLpgSQL_expr *expr)
 		return;
 
 	/*
-	 * OK, we can treat it as a simple plan.
-	 *
-	 * Get the generic plan for the query.  If replanning is needed, do that
-	 * work in the eval_mcontext.
+	 * We have a simple expression.  Although, if it contains inline-able
+	 * SQL functions, better pass it through the planner to get a simpler
+	 * plan.
 	 */
-	oldcontext = MemoryContextSwitchTo(get_eval_mcontext(estate));
-	cplan = SPI_plan_get_cached_plan(expr->plan);
-	MemoryContextSwitchTo(oldcontext);
+	tle_expr = castNode(TargetEntry, linitial(query->targetList))->expr;
+	if (contain_inlinable_functions((Node *) tle_expr))
+	{
+		CachedPlan *cplan = NULL;
+		MemoryContext oldcontext;
+
+		expr->expr_contains_inline_func = true;
+
+		/*
+		 * Get the generic plan for the query.  If replanning is needed, do
+		 * that work in the eval_mcontext.
+		 */
+		oldcontext = MemoryContextSwitchTo(get_eval_mcontext(estate));
+		cplan = SPI_plan_get_cached_plan(expr->plan);
+		MemoryContextSwitchTo(oldcontext);
 
-	/* Can't fail, because we checked for a single CachedPlanSource above */
-	Assert(cplan != NULL);
+		/*
+		 * Can't fail, because we checked for a single CachedPlanSource
+		 * above.
+		 */
+		Assert(cplan != NULL);
 
-	/* Share the remaining work with replan code path */
-	exec_save_simple_expr(expr, cplan);
+		exec_save_simple_expr(expr, cplan);
 
-	/* Release our plan refcount */
-	ReleaseCachedPlan(cplan, true);
+		/* Release our plan refcount */
+		ReleaseCachedPlan(cplan, true);
+	}
+	else
+	{
+		/*
+		 * Save the simple expression, and initialize state to "not valid in
+		 * current transaction".
+		 */
+		expr->expr_simple_expr = tle_expr;
+		expr->expr_simple_generation = plansource->generation;
+		expr->expr_simple_state = NULL;
+		expr->expr_simple_in_use = false;
+		expr->expr_simple_lxid = InvalidLocalTransactionId;
+		/* Also stash away the expression result type */
+		expr->expr_simple_type = exprType((Node *) tle_expr);
+		expr->expr_simple_typmod = exprTypmod((Node *) tle_expr);
+		/* We also want to remember if it is immutable or not */
+		expr->expr_simple_mutable = contain_mutable_functions((Node *) tle_expr);
+		expr->expr_contains_inline_func = false;
+	}
 }
 
 /*
diff --git a/src/pl/plpgsql/src/plpgsql.h b/src/pl/plpgsql/src/plpgsql.h
index b599f67fc5..0e02212135 100644
--- a/src/pl/plpgsql/src/plpgsql.h
+++ b/src/pl/plpgsql/src/plpgsql.h
@@ -245,6 +245,7 @@ typedef struct PLpgSQL_expr
 	ExprState  *expr_simple_state;	/* eval tree for expr_simple_expr */
 	bool		expr_simple_in_use; /* true if eval tree is active */
 	LocalTransactionId expr_simple_lxid;
+	bool		expr_contains_inline_func;
 } PLpgSQL_expr;
 
 /*
diff --git a/src/test/regress/expected/plpgsql.out b/src/test/regress/expected/plpgsql.out
index cd2c79f4d5..05559c927b 100644
--- a/src/test/regress/expected/plpgsql.out
+++ b/src/test/regress/expected/plpgsql.out
@@ -4289,12 +4289,10 @@ end
 $$;
 select fail();
 ERROR:  division by zero
-CONTEXT:  SQL statement "SELECT 1/0"
-PL/pgSQL function fail() line 3 at RETURN
+CONTEXT:  PL/pgSQL function fail() line 3 at RETURN
 select fail();
 ERROR:  division by zero
-CONTEXT:  SQL statement "SELECT 1/0"
-PL/pgSQL function fail() line 3 at RETURN
+CONTEXT:  PL/pgSQL function fail() line 3 at RETURN
 drop function fail();
 -- Test handling of string literals.
 set standard_conforming_strings = off;
@@ -5525,3 +5523,32 @@ END; $$ LANGUAGE plpgsql;
 ERROR:  "x" is not a scalar variable
 LINE 3:   GET DIAGNOSTICS x = ROW_COUNT;
                           ^
+--
+-- Check simple expressions are handled correctly
+--
+DO LANGUAGE plpgsql $$
+DECLARE i bigint = 0;
+BEGIN
+  WHILE i < 100
+  LOOP
+    i := i + 1;
+  END LOOP;
+  RAISE NOTICE '%', i;
+END
+$$;
+NOTICE:  100
+-- Case where the expression is simple but contains inline-able function
+CREATE FUNCTION sql_incr (a bigint) RETURNS bigint AS $$
+SELECT a + 1;
+$$ IMMUTABLE LANGUAGE SQL;
+DO LANGUAGE plpgsql $$
+DECLARE i bigint = 0;
+BEGIN
+  WHILE i < 100
+  LOOP
+    i := sql_incr(i);
+  END LOOP;
+  RAISE NOTICE '%', i;
+END
+$$;
+NOTICE:  100
diff --git a/src/test/regress/sql/plpgsql.sql b/src/test/regress/sql/plpgsql.sql
index d841d8c0f9..bf2bb3c861 100644
--- a/src/test/regress/sql/plpgsql.sql
+++ b/src/test/regress/sql/plpgsql.sql
@@ -4519,3 +4519,33 @@ BEGIN
   GET DIAGNOSTICS x = ROW_COUNT;
   RETURN;
 END; $$ LANGUAGE plpgsql;
+
+--
+-- Check simple expressions are handled correctly
+--
+DO LANGUAGE plpgsql $$
+DECLARE i bigint = 0;
+BEGIN
+  WHILE i < 100
+  LOOP
+    i := i + 1;
+  END LOOP;
+  RAISE NOTICE '%', i;
+END
+$$;
+
+-- Case where the expression is simple but contains inline-able function
+CREATE FUNCTION sql_incr (a bigint) RETURNS bigint AS $$
+SELECT a + 1;
+$$ IMMUTABLE LANGUAGE SQL;
+
+DO LANGUAGE plpgsql $$
+DECLARE i bigint = 0;
+BEGIN
+  WHILE i < 100
+  LOOP
+    i := sql_incr(i);
+  END LOOP;
+  RAISE NOTICE '%', i;
+END
+$$;

diff --git a/src/pl/plpgsql/src/Makefile b/src/pl/plpgsql/src/Makefile
index 70a9c34..193df8a 100644
--- a/src/pl/plpgsql/src/Makefile
+++ b/src/pl/plpgsql/src/Makefile
@@ -33,8 +33,8 @@ DATA = plpgsql.control plpgsql--1.0.sql
 REGRESS_OPTS = --dbname=$(PL_TESTDB)
 
 REGRESS = plpgsql_call plpgsql_control plpgsql_copy plpgsql_domain \
-	plpgsql_record plpgsql_cache plpgsql_transaction plpgsql_trap \
-	plpgsql_trigger plpgsql_varprops
+	plpgsql_record plpgsql_cache plpgsql_simple plpgsql_transaction \
+	plpgsql_trap plpgsql_trigger plpgsql_varprops
 
 # where to find gen_keywordlist.pl and subsidiary files
 TOOLSDIR = $(top_srcdir)/src/tools
diff --git a/src/pl/plpgsql/src/expected/plpgsql_simple.out b/src/pl/plpgsql/src/expected/plpgsql_simple.out
new file mode 100644
index 0000000..5a2fefa
--- /dev/null
+++ b/src/pl/plpgsql/src/expected/plpgsql_simple.out
@@ -0,0 +1,68 @@
+--
+-- Tests for plpgsql's handling of "simple" expressions
+--
+-- Check that changes to an inline-able function are handled correctly
+create function simplesql(int) returns int language sql
+as 'select $1';
+create function simplecaller() returns int language plpgsql
+as $$
+declare
+  sum int := 0;
+begin
+  for n in 1..10 loop
+    sum := sum + simplesql(n);
+    if n = 5 then
+      create or replace function simplesql(int) returns int language sql
+      as 'select $1 + 100';
+    end if;
+  end loop;
+  return sum;
+end$$;
+select simplecaller();
+ simplecaller 
+--------------
+          555
+(1 row)
+
+-- Check that changes in search path are dealt with correctly
+create schema simple1;
+create function simple1.simpletarget(int) returns int language plpgsql
+as $$begin return $1; end$$;
+create function simpletarget(int) returns int language plpgsql
+as $$begin return $1 + 100; end$$;
+create or replace function simplecaller() returns int language plpgsql
+as $$
+declare
+  sum int := 0;
+begin
+  for n in 1..10 loop
+    sum := sum + simpletarget(n);
+    if n = 5 then
+      set local search_path = 'simple1';
+    end if;
+  end loop;
+  return sum;
+end$$;
+select simplecaller();
+ simplecaller 
+--------------
+          555
+(1 row)
+
+-- try it with non-volatile functions, too
+alter function simple1.simpletarget(int) immutable;
+alter function simpletarget(int) immutable;
+select simplecaller();
+ simplecaller 
+--------------
+          555
+(1 row)
+
+-- make sure flushing local caches changes nothing
+\c -
+select simplecaller();
+ simplecaller 
+--------------
+          555
+(1 row)
+
diff --git a/src/pl/plpgsql/src/sql/plpgsql_simple.sql b/src/pl/plpgsql/src/sql/plpgsql_simple.sql
new file mode 100644
index 0000000..8a95768
--- /dev/null
+++ b/src/pl/plpgsql/src/sql/plpgsql_simple.sql
@@ -0,0 +1,61 @@
+--
+-- Tests for plpgsql's handling of "simple" expressions
+--
+
+-- Check that changes to an inline-able function are handled correctly
+create function simplesql(int) returns int language sql
+as 'select $1';
+
+create function simplecaller() returns int language plpgsql
+as $$
+declare
+  sum int := 0;
+begin
+  for n in 1..10 loop
+    sum := sum + simplesql(n);
+    if n = 5 then
+      create or replace function simplesql(int) returns int language sql
+      as 'select $1 + 100';
+    end if;
+  end loop;
+  return sum;
+end$$;
+
+select simplecaller();
+
+
+-- Check that changes in search path are dealt with correctly
+create schema simple1;
+
+create function simple1.simpletarget(int) returns int language plpgsql
+as $$begin return $1; end$$;
+
+create function simpletarget(int) returns int language plpgsql
+as $$begin return $1 + 100; end$$;
+
+create or replace function simplecaller() returns int language plpgsql
+as $$
+declare
+  sum int := 0;
+begin
+  for n in 1..10 loop
+    sum := sum + simpletarget(n);
+    if n = 5 then
+      set local search_path = 'simple1';
+    end if;
+  end loop;
+  return sum;
+end$$;
+
+select simplecaller();
+
+-- try it with non-volatile functions, too
+alter function simple1.simpletarget(int) immutable;
+alter function simpletarget(int) immutable;
+
+select simplecaller();
+
+-- make sure flushing local caches changes nothing
+\c -
+
+select simplecaller();

diff --git a/src/backend/utils/cache/plancache.c b/src/backend/utils/cache/plancache.c
index dbae18d..8e27b03 100644
--- a/src/backend/utils/cache/plancache.c
+++ b/src/backend/utils/cache/plancache.c
@@ -1278,6 +1278,160 @@ ReleaseCachedPlan(CachedPlan *plan, bool useResOwner)
 }
 
 /*
+ * CachedPlanAllowsSimpleValidityCheck: can we use CachedPlanIsSimplyValid?
+ *
+ * This function, together with CachedPlanIsSimplyValid, provides a fast path
+ * for revalidating "simple" generic plans.  The core requirement to be simple
+ * is that the plan must not require taking any locks, which translates to
+ * not touching any tables; this happens to match up well with an important
+ * use-case in PL/pgSQL.  This function tests whether that's true, along
+ * with checking some other corner cases that we'd rather not bother with
+ * handling in the fast path.  (Note that it's still possible for such a plan
+ * to be invalidated, for example due to a change in a function that was
+ * inlined into the plan.)
+ *
+ * This must only be called on known-valid generic plans (eg, ones just
+ * returned by GetCachedPlan).  If it returns true, the caller may re-use
+ * the cached plan as long as CachedPlanIsSimplyValid returns true; that
+ * check is much cheaper than the full revalidation done by GetCachedPlan.
+ * Nonetheless, no required checks are omitted.
+ */
+bool
+CachedPlanAllowsSimpleValidityCheck(CachedPlanSource *plansource,
+									CachedPlan *plan)
+{
+	ListCell   *lc;
+
+	/* Sanity-check that the caller gave us a validated generic plan. */
+	Assert(plansource->magic == CACHEDPLANSOURCE_MAGIC);
+	Assert(plan->magic == CACHEDPLAN_MAGIC);
+	Assert(plansource->is_valid);
+	Assert(plan->is_valid);
+	Assert(plan == plansource->gplan);
+
+	/* We don't support oneshot plans here. */
+	if (plansource->is_oneshot)
+		return false;
+	Assert(!plan->is_oneshot);
+
+	/*
+	 * If the plan is dependent on RLS considerations, or it's transient,
+	 * reject.  These things probably can't ever happen for table-free
+	 * queries, but for safety's sake let's check.
+	 */
+	if (plansource->dependsOnRLS)
+		return false;
+	if (plan->dependsOnRole)
+		return false;
+	if (TransactionIdIsValid(plan->saved_xmin))
+		return false;
+
+	/*
+	 * Reject if AcquirePlannerLocks would have anything to do.  This is
+	 * simplistic, but there's no need to inquire any more carefully; indeed,
+	 * for current callers it shouldn't even be possible to hit any of these
+	 * checks.
+	 */
+	foreach(lc, plansource->query_list)
+	{
+		Query	   *query = lfirst_node(Query, lc);
+
+		if (query->commandType == CMD_UTILITY)
+			return false;
+		if (query->rtable || query->cteList || query->hasSubLinks)
+			return false;
+	}
+
+	/*
+	 * Reject if AcquireExecutorLocks would have anything to do.  This is
+	 * probably unnecessary given the previous check, but let's be safe.
+	 */
+	foreach(lc, plan->stmt_list)
+	{
+		PlannedStmt *plannedstmt = lfirst_node(PlannedStmt, lc);
+		ListCell   *lc2;
+
+		if (plannedstmt->commandType == CMD_UTILITY)
+			return false;
+
+		/*
+		 * We have to grovel through the rtable because it's likely to contain
+		 * an RTE_RESULT relation, rather than being totally empty.
+		 */
+		foreach(lc2, plannedstmt->rtable)
+		{
+			RangeTblEntry *rte = (RangeTblEntry *) lfirst(lc2);
+
+			if (rte->rtekind == RTE_RELATION)
+				return false;
+		}
+	}
+
+	/*
+	 * Okay, it's simple.  Note that what we've primarily established here is
+	 * that no locks need be taken before checking the plan's is_valid flag.
+	 */
+	return true;
+}
+
+/*
+ * CachedPlanIsSimplyValid: quick check for plan still being valid
+ *
+ * This function must not be used unless CachedPlanAllowsSimpleValidityCheck
+ * previously said it was OK.
+ *
+ * If the plan is valid, and "owner" is not NULL, record a refcount on
+ * the plan in that resowner before returning.  It is caller's responsibility
+ * to be sure that a refcount is held on any plan that's being actively used.
+ *
+ * The code here is unconditionally safe as long as the only use of this
+ * CachedPlanSource is in connection with the particular CachedPlan pointer
+ * that's passed in.  If the plansource were being used for other purposes,
+ * it's possible that its generic plan could be invalidated and regenerated
+ * while the current caller wasn't looking, and then there could be a chance
+ * collision of address between this caller's now-stale plan pointer and the
+ * actual address of the new generic plan.  For current uses, that scenario
+ * can't happen; but with a plansource shared across multiple uses, it'd be
+ * advisable to also save plan->generation and verify that that still matches.
+ */
+bool
+CachedPlanIsSimplyValid(CachedPlanSource *plansource, CachedPlan *plan,
+						ResourceOwner owner)
+{
+	/*
+	 * Careful here: since the caller doesn't necessarily hold a refcount on
+	 * the plan to start with, it's possible that "plan" is a dangling
+	 * pointer.  Don't dereference it until we've verified that it still
+	 * matches the plansource's gplan (which is either valid or NULL).
+	 */
+	Assert(plansource->magic == CACHEDPLANSOURCE_MAGIC);
+
+	/*
+	 * Has cache invalidation fired on this plan?  We can check this right
+	 * away since there are no locks that we'd need to acquire first.
+	 */
+	if (!plansource->is_valid || plan != plansource->gplan || !plan->is_valid)
+		return false;
+
+	Assert(plan->magic == CACHEDPLAN_MAGIC);
+
+	/* Is the search_path still the same as when we made it? */
+	Assert(plansource->search_path != NULL);
+	if (!OverrideSearchPathMatchesCurrent(plansource->search_path))
+		return false;
+
+	/* It's still good.  Bump refcount if requested. */
+	if (owner)
+	{
+		ResourceOwnerEnlargePlanCacheRefs(owner);
+		plan->refcount++;
+		ResourceOwnerRememberPlanCacheRef(owner, plan);
+	}
+
+	return true;
+}
+
+/*
  * CachedPlanSetParentContext: move a CachedPlanSource to a new memory context
  *
  * This can only be applied to unsaved plans; once saved, a plan always
diff --git a/src/include/utils/plancache.h b/src/include/utils/plancache.h
index 6a5953c..037edd0 100644
--- a/src/include/utils/plancache.h
+++ b/src/include/utils/plancache.h
@@ -20,6 +20,8 @@
 #include "nodes/params.h"
 #include "tcop/cmdtag.h"
 #include "utils/queryenvironment.h"
+#include "utils/resowner.h"
+
 
 /* Forward declaration, to avoid including parsenodes.h here */
 struct RawStmt;
@@ -220,6 +222,12 @@ extern CachedPlan *GetCachedPlan(CachedPlanSource *plansource,
 								 QueryEnvironment *queryEnv);
 extern void ReleaseCachedPlan(CachedPlan *plan, bool useResOwner);
 
+extern bool CachedPlanAllowsSimpleValidityCheck(CachedPlanSource *plansource,
+												CachedPlan *plan);
+extern bool CachedPlanIsSimplyValid(CachedPlanSource *plansource,
+									CachedPlan *plan,
+									ResourceOwner owner);
+
 extern CachedExpression *GetCachedExpression(Node *expr);
 extern void FreeCachedExpression(CachedExpression *cexpr);
 
diff --git a/src/pl/plpgsql/src/pl_exec.c b/src/pl/plpgsql/src/pl_exec.c
index d3ad4fa..ef33394 100644
--- a/src/pl/plpgsql/src/pl_exec.c
+++ b/src/pl/plpgsql/src/pl_exec.c
@@ -96,6 +96,15 @@ static EState *shared_simple_eval_estate = NULL;
 static SimpleEcontextStackEntry *simple_econtext_stack = NULL;
 
 /*
+ * In addition to the shared simple-eval EState, we have a shared resource
+ * owner that holds refcounts on the CachedPlans for any "simple" expressions
+ * we have evaluated in the current transaction.  This allows us to avoid
+ * continually grabbing and releasing a plan refcount when a simple expression
+ * is used over and over.
+ */
+static ResourceOwner shared_simple_eval_resowner = NULL;
+
+/*
  * Memory management within a plpgsql function generally works with three
  * contexts:
  *
@@ -314,7 +323,8 @@ static int	exec_stmt_set(PLpgSQL_execstate *estate,
 static void plpgsql_estate_setup(PLpgSQL_execstate *estate,
 								 PLpgSQL_function *func,
 								 ReturnSetInfo *rsi,
-								 EState *simple_eval_estate);
+								 EState *simple_eval_estate,
+								 ResourceOwner simple_eval_resowner);
 static void exec_eval_cleanup(PLpgSQL_execstate *estate);
 
 static void exec_prepare_plan(PLpgSQL_execstate *estate,
@@ -440,16 +450,19 @@ static char *format_preparedparamsdata(PLpgSQL_execstate *estate,
  *
  * This is also used to execute inline code blocks (DO blocks).  The only
  * difference that this code is aware of is that for a DO block, we want
- * to use a private simple_eval_estate, which is created and passed in by
- * the caller.  For regular functions, pass NULL, which implies using
- * shared_simple_eval_estate.  (When using a private simple_eval_estate,
+ * to use a private simple_eval_estate and a private simple_eval_resowner,
+ * which are created and passed in by the caller.  For regular functions,
+ * pass NULL, which implies using shared_simple_eval_estate and
+ * shared_simple_eval_resowner.  (When using a private simple_eval_estate,
  * we must also use a private cast hashtable, but that's taken care of
  * within plpgsql_estate_setup.)
  * ----------
  */
 Datum
 plpgsql_exec_function(PLpgSQL_function *func, FunctionCallInfo fcinfo,
-					  EState *simple_eval_estate, bool atomic)
+					  EState *simple_eval_estate,
+					  ResourceOwner simple_eval_resowner,
+					  bool atomic)
 {
 	PLpgSQL_execstate estate;
 	ErrorContextCallback plerrcontext;
@@ -460,7 +473,7 @@ plpgsql_exec_function(PLpgSQL_function *func, FunctionCallInfo fcinfo,
 	 * Setup the execution state
 	 */
 	plpgsql_estate_setup(&estate, func, (ReturnSetInfo *) fcinfo->resultinfo,
-						 simple_eval_estate);
+						 simple_eval_estate, simple_eval_resowner);
 	estate.atomic = atomic;
 
 	/*
@@ -897,7 +910,7 @@ plpgsql_exec_trigger(PLpgSQL_function *func,
 	/*
 	 * Setup the execution state
 	 */
-	plpgsql_estate_setup(&estate, func, NULL, NULL);
+	plpgsql_estate_setup(&estate, func, NULL, NULL, NULL);
 	estate.trigdata = trigdata;
 
 	/*
@@ -1135,7 +1148,7 @@ plpgsql_exec_event_trigger(PLpgSQL_function *func, EventTriggerData *trigdata)
 	/*
 	 * Setup the execution state
 	 */
-	plpgsql_estate_setup(&estate, func, NULL, NULL);
+	plpgsql_estate_setup(&estate, func, NULL, NULL, NULL);
 	estate.evtrigdata = trigdata;
 
 	/*
@@ -2319,6 +2332,7 @@ exec_stmt_call(PLpgSQL_execstate *estate, PLpgSQL_stmt_call *stmt)
 		 * some internal state.
 		 */
 		estate->simple_eval_estate = NULL;
+		estate->simple_eval_resowner = NULL;
 		plpgsql_create_econtext(estate);
 	}
 
@@ -3874,7 +3888,8 @@ static void
 plpgsql_estate_setup(PLpgSQL_execstate *estate,
 					 PLpgSQL_function *func,
 					 ReturnSetInfo *rsi,
-					 EState *simple_eval_estate)
+					 EState *simple_eval_estate,
+					 ResourceOwner simple_eval_resowner)
 {
 	HASHCTL		ctl;
 
@@ -3965,6 +3980,11 @@ plpgsql_estate_setup(PLpgSQL_execstate *estate,
 		estate->cast_hash = shared_cast_hash;
 		estate->cast_hash_context = shared_cast_context;
 	}
+	/* likewise for the simple-expression resource owner */
+	if (simple_eval_resowner)
+		estate->simple_eval_resowner = simple_eval_resowner;
+	else
+		estate->simple_eval_resowner = shared_simple_eval_resowner;
 
 	/*
 	 * We start with no stmt_mcontext; one will be created only if needed.
@@ -4825,6 +4845,7 @@ exec_stmt_commit(PLpgSQL_execstate *estate, PLpgSQL_stmt_commit *stmt)
 	}
 
 	estate->simple_eval_estate = NULL;
+	estate->simple_eval_resowner = NULL;
 	plpgsql_create_econtext(estate);
 
 	return PLPGSQL_RC_OK;
@@ -4847,6 +4868,7 @@ exec_stmt_rollback(PLpgSQL_execstate *estate, PLpgSQL_stmt_rollback *stmt)
 	}
 
 	estate->simple_eval_estate = NULL;
+	estate->simple_eval_resowner = NULL;
 	plpgsql_create_econtext(estate);
 
 	return PLPGSQL_RC_OK;
@@ -6059,8 +6081,6 @@ loop_exit:
  * someone might redefine a SQL function that had been inlined into the simple
  * expression.  That cannot cause a simple expression to become non-simple (or
  * vice versa), but we do have to handle replacing the expression tree.
- * Fortunately it's normally inexpensive to call SPI_plan_get_cached_plan for
- * a simple expression.
  *
  * Note: if pass-by-reference, the result is in the eval_mcontext.
  * It will be freed when exec_eval_cleanup is done.
@@ -6076,7 +6096,7 @@ exec_eval_simple_expr(PLpgSQL_execstate *estate,
 {
 	ExprContext *econtext = estate->eval_econtext;
 	LocalTransactionId curlxid = MyProc->lxid;
-	CachedPlan *cplan;
+	ParamListInfo paramLI;
 	void	   *save_setup_arg;
 	bool		need_snapshot;
 	MemoryContext oldcontext;
@@ -6090,29 +6110,92 @@ exec_eval_simple_expr(PLpgSQL_execstate *estate,
 	/*
 	 * If expression is in use in current xact, don't touch it.
 	 */
-	if (expr->expr_simple_in_use && expr->expr_simple_lxid == curlxid)
+	if (unlikely(expr->expr_simple_in_use) &&
+		expr->expr_simple_lxid == curlxid)
 		return false;
 
 	/*
-	 * Revalidate cached plan, so that we will notice if it became stale. (We
-	 * need to hold a refcount while using the plan, anyway.)  If replanning
-	 * is needed, do that work in the eval_mcontext.
+	 * Check to see if the cached plan has been invalidated.  If not, and this
+	 * is the first use in the current transaction, save a plan refcount in
+	 * the simple-expression resowner.
 	 */
-	oldcontext = MemoryContextSwitchTo(get_eval_mcontext(estate));
-	cplan = SPI_plan_get_cached_plan(expr->plan);
-	MemoryContextSwitchTo(oldcontext);
+	if (likely(CachedPlanIsSimplyValid(expr->expr_simple_plansource,
+									   expr->expr_simple_plan,
+									   (expr->expr_simple_plan_lxid != curlxid ?
+										estate->simple_eval_resowner : NULL))))
+	{
+		/*
+		 * It's still good, so just remember that we have a refcount on the
+		 * plan in the current transaction.  (If we already had one, this
+		 * assignment is a no-op.)
+		 */
+		expr->expr_simple_plan_lxid = curlxid;
+	}
+	else
+	{
+		/* Need to replan */
+		CachedPlan *cplan;
 
-	/*
-	 * We can't get a failure here, because the number of CachedPlanSources in
-	 * the SPI plan can't change from what exec_simple_check_plan saw; it's a
-	 * property of the raw parsetree generated from the query text.
-	 */
-	Assert(cplan != NULL);
+		/*
+		 * If we have a valid refcount on some previous version of the plan,
+		 * release it, so we don't leak plans intra-transaction.
+		 */
+		if (expr->expr_simple_plan_lxid == curlxid)
+		{
+			ResourceOwner saveResourceOwner = CurrentResourceOwner;
 
-	/* If it got replanned, update our copy of the simple expression */
-	if (cplan->generation != expr->expr_simple_generation)
-	{
+			CurrentResourceOwner = estate->simple_eval_resowner;
+			ReleaseCachedPlan(expr->expr_simple_plan, true);
+			CurrentResourceOwner = saveResourceOwner;
+			expr->expr_simple_plan = NULL;
+			expr->expr_simple_plan_lxid = InvalidLocalTransactionId;
+		}
+
+		/* Do the replanning work in the eval_mcontext */
+		oldcontext = MemoryContextSwitchTo(get_eval_mcontext(estate));
+		cplan = SPI_plan_get_cached_plan(expr->plan);
+		MemoryContextSwitchTo(oldcontext);
+
+		/*
+		 * We can't get a failure here, because the number of
+		 * CachedPlanSources in the SPI plan can't change from what
+		 * exec_simple_check_plan saw; it's a property of the raw parsetree
+		 * generated from the query text.
+		 */
+		Assert(cplan != NULL);
+
+		/*
+		 * These tests probably can't fail either, but if they do, cope by
+		 * declaring the plan to be non-simple.  On success, we'll acquire a
+		 * refcount on the new plan, stored in simple_eval_resowner.
+		 */
+		if (CachedPlanAllowsSimpleValidityCheck(expr->expr_simple_plansource,
+												cplan) &&
+			CachedPlanIsSimplyValid(expr->expr_simple_plansource, cplan,
+									estate->simple_eval_resowner))
+		{
+			/* Remember that we have the refcount */
+			expr->expr_simple_plan = cplan;
+			expr->expr_simple_plan_lxid = curlxid;
+		}
+		else
+		{
+			/* Release SPI_plan_get_cached_plan's refcount */
+			ReleaseCachedPlan(cplan, true);
+			/* Mark expression as non-simple, and fail */
+			expr->expr_simple_expr = NULL;
+			return false;
+		}
+
+		/*
+		 * SPI_plan_get_cached_plan acquired a plan refcount stored in the
+		 * active resowner.  We don't need that anymore, so release it.
+		 */
+		ReleaseCachedPlan(cplan, true);
+
+		/* Extract desired scalar expression from cached plan */
 		exec_save_simple_expr(expr, cplan);
+
 		/* better recheck r/w safety, as it could change due to inlining */
 		if (expr->rwparam >= 0)
 			exec_check_rw_parameter(expr, expr->rwparam);
@@ -6128,16 +6211,24 @@ exec_eval_simple_expr(PLpgSQL_execstate *estate,
 	 * Set up ParamListInfo to pass to executor.  For safety, save and restore
 	 * estate->paramLI->parserSetupArg around our use of the param list.
 	 */
-	save_setup_arg = estate->paramLI->parserSetupArg;
+	paramLI = estate->paramLI;
+	save_setup_arg = paramLI->parserSetupArg;
 
-	econtext->ecxt_param_list_info = setup_param_list(estate, expr);
+	/*
+	 * We can skip using setup_param_list() in favor of just doing this
+	 * unconditionally, because there's no need for the optimization of
+	 * possibly setting ecxt_param_list_info to NULL; we've already forced use
+	 * of a generic plan.
+	 */
+	paramLI->parserSetupArg = (void *) expr;
+	econtext->ecxt_param_list_info = paramLI;
 
 	/*
 	 * Prepare the expression for execution, if it's not been done already in
 	 * the current transaction.  (This will be forced to happen if we called
 	 * exec_save_simple_expr above.)
 	 */
-	if (expr->expr_simple_lxid != curlxid)
+	if (unlikely(expr->expr_simple_lxid != curlxid))
 	{
 		oldcontext = MemoryContextSwitchTo(estate->simple_eval_estate->es_query_cxt);
 		expr->expr_simple_state =
@@ -6185,7 +6276,7 @@ exec_eval_simple_expr(PLpgSQL_execstate *estate,
 
 	econtext->ecxt_param_list_info = NULL;
 
-	estate->paramLI->parserSetupArg = save_setup_arg;
+	paramLI->parserSetupArg = save_setup_arg;
 
 	if (need_snapshot)
 		PopActiveSnapshot();
@@ -6193,11 +6284,6 @@ exec_eval_simple_expr(PLpgSQL_execstate *estate,
 	MemoryContextSwitchTo(oldcontext);
 
 	/*
-	 * Now we can release our refcount on the cached plan.
-	 */
-	ReleaseCachedPlan(cplan, true);
-
-	/*
 	 * That's it.
 	 */
 	return true;
@@ -7984,10 +8070,35 @@ exec_simple_check_plan(PLpgSQL_execstate *estate, PLpgSQL_expr *expr)
 	/* Can't fail, because we checked for a single CachedPlanSource above */
 	Assert(cplan != NULL);
 
-	/* Share the remaining work with replan code path */
-	exec_save_simple_expr(expr, cplan);
+	/*
+	 * Verify that plancache.c thinks the plan is simple enough to use
+	 * CachedPlanIsSimplyValid.  Given the restrictions above, it's unlikely
+	 * that this could fail, but if it does, just treat plan as not simple.
+	 */
+	if (CachedPlanAllowsSimpleValidityCheck(plansource, cplan))
+	{
+		/*
+		 * OK, use CachedPlanIsSimplyValid to save a refcount on the plan in
+		 * the simple-expression resowner.  This shouldn't fail either, but if
+		 * somehow it does, again we can cope by treating plan as not simple.
+		 */
+		if (CachedPlanIsSimplyValid(plansource, cplan,
+									estate->simple_eval_resowner))
+		{
+			/* Remember that we have the refcount */
+			expr->expr_simple_plansource = plansource;
+			expr->expr_simple_plan = cplan;
+			expr->expr_simple_plan_lxid = MyProc->lxid;
+
+			/* Share the remaining work with the replan code path */
+			exec_save_simple_expr(expr, cplan);
+		}
+	}
 
-	/* Release our plan refcount */
+	/*
+	 * Release the plan refcount obtained by SPI_plan_get_cached_plan.  (This
+	 * refcount is held by the wrong resowner, so we can't just repurpose it.)
+	 */
 	ReleaseCachedPlan(cplan, true);
 }
 
@@ -8060,7 +8171,6 @@ exec_save_simple_expr(PLpgSQL_expr *expr, CachedPlan *cplan)
 	 * current transaction".
 	 */
 	expr->expr_simple_expr = tle_expr;
-	expr->expr_simple_generation = cplan->generation;
 	expr->expr_simple_state = NULL;
 	expr->expr_simple_in_use = false;
 	expr->expr_simple_lxid = InvalidLocalTransactionId;
@@ -8196,7 +8306,7 @@ exec_set_found(PLpgSQL_execstate *estate, bool state)
  *
  * We may need to create a new shared_simple_eval_estate too, if there's not
  * one already for the current transaction.  The EState will be cleaned up at
- * transaction end.
+ * transaction end.  Ditto for shared_simple_eval_resowner.
  */
 static void
 plpgsql_create_econtext(PLpgSQL_execstate *estate)
@@ -8225,6 +8335,20 @@ plpgsql_create_econtext(PLpgSQL_execstate *estate)
 	}
 
 	/*
+	 * Likewise for the simple-expression resource owner.  (Note: it'd be
+	 * safer to create this as a child of TopTransactionResourceOwner; but
+	 * right now that causes issues in transaction-spanning procedures, so
+	 * make it standalone.)
+	 */
+	if (estate->simple_eval_resowner == NULL)
+	{
+		if (shared_simple_eval_resowner == NULL)
+			shared_simple_eval_resowner =
+				ResourceOwnerCreate(NULL, "PL/pgSQL simple expressions");
+		estate->simple_eval_resowner = shared_simple_eval_resowner;
+	}
+
+	/*
 	 * Create a child econtext for the current function.
 	 */
 	estate->eval_econtext = CreateExprContext(estate->simple_eval_estate);
@@ -8270,16 +8394,17 @@ plpgsql_destroy_econtext(PLpgSQL_execstate *estate)
  * plpgsql_xact_cb --- post-transaction-commit-or-abort cleanup
  *
  * If a simple-expression EState was created in the current transaction,
- * it has to be cleaned up.
+ * it has to be cleaned up.  The same for the simple-expression resowner.
  */
 void
 plpgsql_xact_cb(XactEvent event, void *arg)
 {
 	/*
 	 * If we are doing a clean transaction shutdown, free the EState (so that
-	 * any remaining resources will be released correctly). In an abort, we
+	 * any remaining resources will be released correctly).  In an abort, we
 	 * expect the regular abort recovery procedures to release everything of
-	 * interest.
+	 * interest.  The resowner has to be explicitly released in both cases,
+	 * though, since it's not a child of TopTransactionResourceOwner.
 	 */
 	if (event == XACT_EVENT_COMMIT || event == XACT_EVENT_PREPARE)
 	{
@@ -8288,11 +8413,17 @@ plpgsql_xact_cb(XactEvent event, void *arg)
 		if (shared_simple_eval_estate)
 			FreeExecutorState(shared_simple_eval_estate);
 		shared_simple_eval_estate = NULL;
+		if (shared_simple_eval_resowner)
+			plpgsql_free_simple_resowner(shared_simple_eval_resowner);
+		shared_simple_eval_resowner = NULL;
 	}
 	else if (event == XACT_EVENT_ABORT)
 	{
 		simple_econtext_stack = NULL;
 		shared_simple_eval_estate = NULL;
+		if (shared_simple_eval_resowner)
+			plpgsql_free_simple_resowner(shared_simple_eval_resowner);
+		shared_simple_eval_resowner = NULL;
 	}
 }
 
@@ -8324,6 +8455,33 @@ plpgsql_subxact_cb(SubXactEvent event, SubTransactionId mySubid,
 }
 
 /*
+ * plpgsql_free_simple_resowner --- free a simple-expression ResourceOwner
+ */
+void
+plpgsql_free_simple_resowner(ResourceOwner simple_eval_resowner)
+{
+	/*
+	 * At this writing, the only thing that could actually get released is
+	 * plancache refcounts; but we may as well do the full release protocol.
+	 *
+	 * We pass isCommit = false even when committing, to suppress
+	 * resource-leakage gripes, since we aren't bothering to release the
+	 * refcounts one-by-one.
+	 */
+	ResourceOwnerRelease(simple_eval_resowner,
+						 RESOURCE_RELEASE_BEFORE_LOCKS,
+						 false, true);
+	ResourceOwnerRelease(simple_eval_resowner,
+						 RESOURCE_RELEASE_LOCKS,
+						 false, true);
+	ResourceOwnerRelease(simple_eval_resowner,
+						 RESOURCE_RELEASE_AFTER_LOCKS,
+						 false, true);
+
+	ResourceOwnerDelete(simple_eval_resowner);
+}
+
+/*
  * assign_simple_var --- assign a new value to any VAR datum.
  *
  * This should be the only mechanism for assignment to simple variables,
diff --git a/src/pl/plpgsql/src/pl_handler.c b/src/pl/plpgsql/src/pl_handler.c
index b434818..4cc6079 100644
--- a/src/pl/plpgsql/src/pl_handler.c
+++ b/src/pl/plpgsql/src/pl_handler.c
@@ -262,7 +262,9 @@ plpgsql_call_handler(PG_FUNCTION_ARGS)
 			retval = (Datum) 0;
 		}
 		else
-			retval = plpgsql_exec_function(func, fcinfo, NULL, !nonatomic);
+			retval = plpgsql_exec_function(func, fcinfo,
+										   NULL, NULL,
+										   !nonatomic);
 	}
 	PG_FINALLY();
 	{
@@ -297,6 +299,7 @@ plpgsql_inline_handler(PG_FUNCTION_ARGS)
 	PLpgSQL_function *func;
 	FmgrInfo	flinfo;
 	EState	   *simple_eval_estate;
+	ResourceOwner simple_eval_resowner;
 	Datum		retval;
 	int			rc;
 
@@ -323,22 +326,27 @@ plpgsql_inline_handler(PG_FUNCTION_ARGS)
 	flinfo.fn_oid = InvalidOid;
 	flinfo.fn_mcxt = CurrentMemoryContext;
 
-	/* Create a private EState for simple-expression execution */
+	/* Create a private EState and resowner for simple-expression execution */
 	simple_eval_estate = CreateExecutorState();
+	simple_eval_resowner =
+		ResourceOwnerCreate(NULL, "PL/pgSQL DO block simple expressions");
 
 	/* And run the function */
 	PG_TRY();
 	{
-		retval = plpgsql_exec_function(func, fake_fcinfo, simple_eval_estate, codeblock->atomic);
+		retval = plpgsql_exec_function(func, fake_fcinfo,
+									   simple_eval_estate,
+									   simple_eval_resowner,
+									   codeblock->atomic);
 	}
 	PG_CATCH();
 	{
 		/*
 		 * We need to clean up what would otherwise be long-lived resources
 		 * accumulated by the failed DO block, principally cached plans for
-		 * statements (which can be flushed with plpgsql_free_function_memory)
-		 * and execution trees for simple expressions, which are in the
-		 * private EState.
+		 * statements (which can be flushed by plpgsql_free_function_memory),
+		 * execution trees for simple expressions, which are in the private
+		 * EState, and cached-plan refcounts held by the private resowner.
 		 *
 		 * Before releasing the private EState, we must clean up any
 		 * simple_econtext_stack entries pointing into it, which we can do by
@@ -351,8 +359,9 @@ plpgsql_inline_handler(PG_FUNCTION_ARGS)
 						   GetCurrentSubTransactionId(),
 						   0, NULL);
 
-		/* Clean up the private EState */
+		/* Clean up the private EState and resowner */
 		FreeExecutorState(simple_eval_estate);
+		plpgsql_free_simple_resowner(simple_eval_resowner);
 
 		/* Function should now have no remaining use-counts ... */
 		func->use_count--;
@@ -366,8 +375,9 @@ plpgsql_inline_handler(PG_FUNCTION_ARGS)
 	}
 	PG_END_TRY();
 
-	/* Clean up the private EState */
+	/* Clean up the private EState and resowner */
 	FreeExecutorState(simple_eval_estate);
+	plpgsql_free_simple_resowner(simple_eval_resowner);
 
 	/* Function should now have no remaining use-counts ... */
 	func->use_count--;
diff --git a/src/pl/plpgsql/src/plpgsql.h b/src/pl/plpgsql/src/plpgsql.h
index 69df330..410dcde 100644
--- a/src/pl/plpgsql/src/plpgsql.h
+++ b/src/pl/plpgsql/src/plpgsql.h
@@ -231,12 +231,21 @@ typedef struct PLpgSQL_expr
 
 	/* fields for "simple expression" fast-path execution: */
 	Expr	   *expr_simple_expr;	/* NULL means not a simple expr */
-	int			expr_simple_generation; /* plancache generation we checked */
 	Oid			expr_simple_type;	/* result type Oid, if simple */
 	int32		expr_simple_typmod; /* result typmod, if simple */
 	bool		expr_simple_mutable;	/* true if simple expr is mutable */
 
 	/*
+	 * If the expression was ever determined to be simple, we remember its
+	 * CachedPlanSource and CachedPlan here.  If expr_simple_plan_lxid matches
+	 * current LXID, then we hold a refcount on expr_simple_plan in the
+	 * current transaction.  Otherwise we need to get one before re-using it.
+	 */
+	CachedPlanSource *expr_simple_plansource;	/* extracted from "plan" */
+	CachedPlan *expr_simple_plan;	/* extracted from "plan" */
+	LocalTransactionId expr_simple_plan_lxid;
+
+	/*
 	 * if expr is simple AND prepared in current transaction,
 	 * expr_simple_state and expr_simple_in_use are valid. Test validity by
 	 * seeing if expr_simple_lxid matches current LXID.  (If not,
@@ -1082,8 +1091,9 @@ typedef struct PLpgSQL_execstate
 	 */
 	ParamListInfo paramLI;
 
-	/* EState to use for "simple" expression evaluation */
+	/* EState and resowner to use for "simple" expression evaluation */
 	EState	   *simple_eval_estate;
+	ResourceOwner simple_eval_resowner;
 
 	/* lookup table to use for executing type casts */
 	HTAB	   *cast_hash;
@@ -1268,6 +1278,7 @@ extern void _PG_init(void);
 extern Datum plpgsql_exec_function(PLpgSQL_function *func,
 								   FunctionCallInfo fcinfo,
 								   EState *simple_eval_estate,
+								   ResourceOwner simple_eval_resowner,
 								   bool atomic);
 extern HeapTuple plpgsql_exec_trigger(PLpgSQL_function *func,
 									  TriggerData *trigdata);
@@ -1276,6 +1287,7 @@ extern void plpgsql_exec_event_trigger(PLpgSQL_function *func,
 extern void plpgsql_xact_cb(XactEvent event, void *arg);
 extern void plpgsql_subxact_cb(SubXactEvent event, SubTransactionId mySubid,
 							   SubTransactionId parentSubid, void *arg);
+extern void plpgsql_free_simple_resowner(ResourceOwner simple_eval_resowner);
 extern Oid	plpgsql_exec_get_datum_type(PLpgSQL_execstate *estate,
 										PLpgSQL_datum *datum);
 extern void plpgsql_exec_get_datum_type_info(PLpgSQL_execstate *estate,

diff --git a/src/backend/catalog/namespace.c b/src/backend/catalog/namespace.c
index 5ff7824..2ec2301 100644
--- a/src/backend/catalog/namespace.c
+++ b/src/backend/catalog/namespace.c
@@ -126,6 +126,11 @@
  * namespaceUser is the userid the path has been computed for.
  *
  * Note: all data pointed to by these List variables is in TopMemoryContext.
+ *
+ * activePathGeneration is incremented whenever the effective values of
+ * activeSearchPath/activeCreationNamespace/activeTempCreationPending change.
+ * This can be used to quickly detect whether any change has happened since
+ * a previous examination of the search path state.
  */
 
 /* These variables define the actually active state: */
@@ -138,6 +143,9 @@ static Oid	activeCreationNamespace = InvalidOid;
 /* if true, activeCreationNamespace is wrong, it should be temp namespace */
 static bool activeTempCreationPending = false;
 
+/* current generation counter; make sure this is never zero */
+static uint64 activePathGeneration = 1;
+
 /* These variables are the values last derived from namespace_search_path: */
 
 static List *baseSearchPath = NIL;
@@ -3373,6 +3381,7 @@ GetOverrideSearchPath(MemoryContext context)
 		schemas = list_delete_first(schemas);
 	}
 	result->schemas = schemas;
+	result->generation = activePathGeneration;
 
 	MemoryContextSwitchTo(oldcxt);
 
@@ -3393,12 +3402,18 @@ CopyOverrideSearchPath(OverrideSearchPath *path)
 	result->schemas = list_copy(path->schemas);
 	result->addCatalog = path->addCatalog;
 	result->addTemp = path->addTemp;
+	result->generation = path->generation;
 
 	return result;
 }
 
 /*
  * OverrideSearchPathMatchesCurrent - does path match current setting?
+ *
+ * This is tested over and over in some common code paths, and in the typical
+ * scenario where the active search path seldom changes, it'll always succeed.
+ * We make that case fast by keeping a generation counter that is advanced
+ * whenever the active search path changes.
  */
 bool
 OverrideSearchPathMatchesCurrent(OverrideSearchPath *path)
@@ -3408,6 +3423,10 @@ OverrideSearchPathMatchesCurrent(OverrideSearchPath *path)
 
 	recomputeNamespacePath();
 
+	/* Quick out if already known equal to active path. */
+	if (path->generation == activePathGeneration)
+		return true;
+
 	/* We scan down the activeSearchPath to see if it matches the input. */
 	lc = list_head(activeSearchPath);
 
@@ -3440,6 +3459,13 @@ OverrideSearchPathMatchesCurrent(OverrideSearchPath *path)
 	}
 	if (lc)
 		return false;
+
+	/*
+	 * Update path->generation so that future tests will return quickly, so
+	 * long as the active search path doesn't change.
+	 */
+	path->generation = activePathGeneration;
+
 	return true;
 }
 
@@ -3510,6 +3536,14 @@ PushOverrideSearchPath(OverrideSearchPath *newpath)
 	activeCreationNamespace = entry->creationNamespace;
 	activeTempCreationPending = false;	/* XXX is this OK? */
 
+	/*
+	 * We always increment activePathGeneration when pushing/popping an
+	 * override path.  In current usage, these actions always change the
+	 * effective path state, so there's no value in checking to see if it
+	 * didn't change.
+	 */
+	activePathGeneration++;
+
 	MemoryContextSwitchTo(oldcxt);
 }
 
@@ -3551,6 +3585,9 @@ PopOverrideSearchPath(void)
 		activeCreationNamespace = baseCreationNamespace;
 		activeTempCreationPending = baseTempCreationPending;
 	}
+
+	/* As above, the generation always increments. */
+	activePathGeneration++;
 }
 
 
@@ -3707,6 +3744,7 @@ recomputeNamespacePath(void)
 	ListCell   *l;
 	bool		temp_missing;
 	Oid			firstNS;
+	bool		pathChanged;
 	MemoryContext oldcxt;
 
 	/* Do nothing if an override search spec is active. */
@@ -3814,18 +3852,31 @@ recomputeNamespacePath(void)
 		oidlist = lcons_oid(myTempNamespace, oidlist);
 
 	/*
-	 * Now that we've successfully built the new list of namespace OIDs, save
-	 * it in permanent storage.
+	 * We want to detect the case where the effective value of the base search
+	 * path variables didn't change.  As long as we're doing so, we can avoid
+	 * copying the OID list unncessarily.
 	 */
-	oldcxt = MemoryContextSwitchTo(TopMemoryContext);
-	newpath = list_copy(oidlist);
-	MemoryContextSwitchTo(oldcxt);
+	if (baseCreationNamespace == firstNS &&
+		baseTempCreationPending == temp_missing &&
+		equal(oidlist, baseSearchPath))
+	{
+		pathChanged = false;
+	}
+	else
+	{
+		pathChanged = true;
+
+		/* Must save OID list in permanent storage. */
+		oldcxt = MemoryContextSwitchTo(TopMemoryContext);
+		newpath = list_copy(oidlist);
+		MemoryContextSwitchTo(oldcxt);
 
-	/* Now safe to assign to state variables. */
-	list_free(baseSearchPath);
-	baseSearchPath = newpath;
-	baseCreationNamespace = firstNS;
-	baseTempCreationPending = temp_missing;
+		/* Now safe to assign to state variables. */
+		list_free(baseSearchPath);
+		baseSearchPath = newpath;
+		baseCreationNamespace = firstNS;
+		baseTempCreationPending = temp_missing;
+	}
 
 	/* Mark the path valid. */
 	baseSearchPathValid = true;
@@ -3836,6 +3887,16 @@ recomputeNamespacePath(void)
 	activeCreationNamespace = baseCreationNamespace;
 	activeTempCreationPending = baseTempCreationPending;
 
+	/*
+	 * Bump the generation only if something actually changed.  (Notice that
+	 * what we compared to was the old state of the base path variables; so
+	 * this does not deal with the situation where we have just popped an
+	 * override path and restored the prior state of the base path.  Instead
+	 * we rely on the override-popping logic to have bumped the generation.)
+	 */
+	if (pathChanged)
+		activePathGeneration++;
+
 	/* Clean up. */
 	pfree(rawname);
 	list_free(namelist);
@@ -4054,6 +4115,8 @@ AtEOXact_Namespace(bool isCommit, bool parallel)
 		activeSearchPath = baseSearchPath;
 		activeCreationNamespace = baseCreationNamespace;
 		activeTempCreationPending = baseTempCreationPending;
+		/* Always bump generation --- see note in recomputeNamespacePath */
+		activePathGeneration++;
 	}
 }
 
@@ -4109,6 +4172,8 @@ AtEOSubXact_Namespace(bool isCommit, SubTransactionId mySubid,
 		overrideStack = list_delete_first(overrideStack);
 		list_free(entry->searchPath);
 		pfree(entry);
+		/* Always bump generation --- see note in recomputeNamespacePath */
+		activePathGeneration++;
 	}
 
 	/* Activate the next level down. */
@@ -4118,6 +4183,12 @@ AtEOSubXact_Namespace(bool isCommit, SubTransactionId mySubid,
 		activeSearchPath = entry->searchPath;
 		activeCreationNamespace = entry->creationNamespace;
 		activeTempCreationPending = false;	/* XXX is this OK? */
+
+		/*
+		 * It's probably unnecessary to bump generation here, but this should
+		 * not be a performance-critical case, so better to be over-cautious.
+		 */
+		activePathGeneration++;
 	}
 	else
 	{
@@ -4125,6 +4196,12 @@ AtEOSubXact_Namespace(bool isCommit, SubTransactionId mySubid,
 		activeSearchPath = baseSearchPath;
 		activeCreationNamespace = baseCreationNamespace;
 		activeTempCreationPending = baseTempCreationPending;
+
+		/*
+		 * If we popped an override stack entry, then we already bumped the
+		 * generation above.  If we did not, then the above assignments did
+		 * nothing and we need not bump the generation.
+		 */
 	}
 }
 
@@ -4264,6 +4341,7 @@ InitializeSearchPath(void)
 		activeSearchPath = baseSearchPath;
 		activeCreationNamespace = baseCreationNamespace;
 		activeTempCreationPending = baseTempCreationPending;
+		activePathGeneration++; /* pro forma */
 	}
 	else
 	{
diff --git a/src/include/catalog/namespace.h b/src/include/catalog/namespace.h
index 3e3a192..2456c08 100644
--- a/src/include/catalog/namespace.h
+++ b/src/include/catalog/namespace.h
@@ -49,12 +49,17 @@ typedef enum TempNamespaceStatus
 
 /*
  *	Structure for xxxOverrideSearchPath functions
+ *
+ * The generation counter is private to namespace.c and shouldn't be touched
+ * by other code.  It can be initialized to zero if necessary (that means
+ * "not known equal to the current active path").
  */
 typedef struct OverrideSearchPath
 {
 	List	   *schemas;		/* OIDs of explicitly named schemas */
 	bool		addCatalog;		/* implicitly prepend pg_catalog? */
 	bool		addTemp;		/* implicitly prepend temp schema? */
+	uint64		generation;		/* for quick detection of equality to active */
 } OverrideSearchPath;
 
 /*

diff --git a/src/pl/plpgsql/src/Makefile b/src/pl/plpgsql/src/Makefile
index 70a9c34..193df8a 100644
--- a/src/pl/plpgsql/src/Makefile
+++ b/src/pl/plpgsql/src/Makefile
@@ -33,8 +33,8 @@ DATA = plpgsql.control plpgsql--1.0.sql
 REGRESS_OPTS = --dbname=$(PL_TESTDB)
 
 REGRESS = plpgsql_call plpgsql_control plpgsql_copy plpgsql_domain \
-	plpgsql_record plpgsql_cache plpgsql_transaction plpgsql_trap \
-	plpgsql_trigger plpgsql_varprops
+	plpgsql_record plpgsql_cache plpgsql_simple plpgsql_transaction \
+	plpgsql_trap plpgsql_trigger plpgsql_varprops
 
 # where to find gen_keywordlist.pl and subsidiary files
 TOOLSDIR = $(top_srcdir)/src/tools
diff --git a/src/pl/plpgsql/src/expected/plpgsql_simple.out b/src/pl/plpgsql/src/expected/plpgsql_simple.out
new file mode 100644
index 0000000..5a2fefa
--- /dev/null
+++ b/src/pl/plpgsql/src/expected/plpgsql_simple.out
@@ -0,0 +1,68 @@
+--
+-- Tests for plpgsql's handling of "simple" expressions
+--
+-- Check that changes to an inline-able function are handled correctly
+create function simplesql(int) returns int language sql
+as 'select $1';
+create function simplecaller() returns int language plpgsql
+as $$
+declare
+  sum int := 0;
+begin
+  for n in 1..10 loop
+    sum := sum + simplesql(n);
+    if n = 5 then
+      create or replace function simplesql(int) returns int language sql
+      as 'select $1 + 100';
+    end if;
+  end loop;
+  return sum;
+end$$;
+select simplecaller();
+ simplecaller 
+--------------
+          555
+(1 row)
+
+-- Check that changes in search path are dealt with correctly
+create schema simple1;
+create function simple1.simpletarget(int) returns int language plpgsql
+as $$begin return $1; end$$;
+create function simpletarget(int) returns int language plpgsql
+as $$begin return $1 + 100; end$$;
+create or replace function simplecaller() returns int language plpgsql
+as $$
+declare
+  sum int := 0;
+begin
+  for n in 1..10 loop
+    sum := sum + simpletarget(n);
+    if n = 5 then
+      set local search_path = 'simple1';
+    end if;
+  end loop;
+  return sum;
+end$$;
+select simplecaller();
+ simplecaller 
+--------------
+          555
+(1 row)
+
+-- try it with non-volatile functions, too
+alter function simple1.simpletarget(int) immutable;
+alter function simpletarget(int) immutable;
+select simplecaller();
+ simplecaller 
+--------------
+          555
+(1 row)
+
+-- make sure flushing local caches changes nothing
+\c -
+select simplecaller();
+ simplecaller 
+--------------
+          555
+(1 row)
+
diff --git a/src/pl/plpgsql/src/sql/plpgsql_simple.sql b/src/pl/plpgsql/src/sql/plpgsql_simple.sql
new file mode 100644
index 0000000..8a95768
--- /dev/null
+++ b/src/pl/plpgsql/src/sql/plpgsql_simple.sql
@@ -0,0 +1,61 @@
+--
+-- Tests for plpgsql's handling of "simple" expressions
+--
+
+-- Check that changes to an inline-able function are handled correctly
+create function simplesql(int) returns int language sql
+as 'select $1';
+
+create function simplecaller() returns int language plpgsql
+as $$
+declare
+  sum int := 0;
+begin
+  for n in 1..10 loop
+    sum := sum + simplesql(n);
+    if n = 5 then
+      create or replace function simplesql(int) returns int language sql
+      as 'select $1 + 100';
+    end if;
+  end loop;
+  return sum;
+end$$;
+
+select simplecaller();
+
+
+-- Check that changes in search path are dealt with correctly
+create schema simple1;
+
+create function simple1.simpletarget(int) returns int language plpgsql
+as $$begin return $1; end$$;
+
+create function simpletarget(int) returns int language plpgsql
+as $$begin return $1 + 100; end$$;
+
+create or replace function simplecaller() returns int language plpgsql
+as $$
+declare
+  sum int := 0;
+begin
+  for n in 1..10 loop
+    sum := sum + simpletarget(n);
+    if n = 5 then
+      set local search_path = 'simple1';
+    end if;
+  end loop;
+  return sum;
+end$$;
+
+select simplecaller();
+
+-- try it with non-volatile functions, too
+alter function simple1.simpletarget(int) immutable;
+alter function simpletarget(int) immutable;
+
+select simplecaller();
+
+-- make sure flushing local caches changes nothing
+\c -
+
+select simplecaller();

diff --git a/src/backend/utils/cache/plancache.c b/src/backend/utils/cache/plancache.c
index dbae18d..8e27b03 100644
--- a/src/backend/utils/cache/plancache.c
+++ b/src/backend/utils/cache/plancache.c
@@ -1278,6 +1278,160 @@ ReleaseCachedPlan(CachedPlan *plan, bool useResOwner)
 }
 
 /*
+ * CachedPlanAllowsSimpleValidityCheck: can we use CachedPlanIsSimplyValid?
+ *
+ * This function, together with CachedPlanIsSimplyValid, provides a fast path
+ * for revalidating "simple" generic plans.  The core requirement to be simple
+ * is that the plan must not require taking any locks, which translates to
+ * not touching any tables; this happens to match up well with an important
+ * use-case in PL/pgSQL.  This function tests whether that's true, along
+ * with checking some other corner cases that we'd rather not bother with
+ * handling in the fast path.  (Note that it's still possible for such a plan
+ * to be invalidated, for example due to a change in a function that was
+ * inlined into the plan.)
+ *
+ * This must only be called on known-valid generic plans (eg, ones just
+ * returned by GetCachedPlan).  If it returns true, the caller may re-use
+ * the cached plan as long as CachedPlanIsSimplyValid returns true; that
+ * check is much cheaper than the full revalidation done by GetCachedPlan.
+ * Nonetheless, no required checks are omitted.
+ */
+bool
+CachedPlanAllowsSimpleValidityCheck(CachedPlanSource *plansource,
+									CachedPlan *plan)
+{
+	ListCell   *lc;
+
+	/* Sanity-check that the caller gave us a validated generic plan. */
+	Assert(plansource->magic == CACHEDPLANSOURCE_MAGIC);
+	Assert(plan->magic == CACHEDPLAN_MAGIC);
+	Assert(plansource->is_valid);
+	Assert(plan->is_valid);
+	Assert(plan == plansource->gplan);
+
+	/* We don't support oneshot plans here. */
+	if (plansource->is_oneshot)
+		return false;
+	Assert(!plan->is_oneshot);
+
+	/*
+	 * If the plan is dependent on RLS considerations, or it's transient,
+	 * reject.  These things probably can't ever happen for table-free
+	 * queries, but for safety's sake let's check.
+	 */
+	if (plansource->dependsOnRLS)
+		return false;
+	if (plan->dependsOnRole)
+		return false;
+	if (TransactionIdIsValid(plan->saved_xmin))
+		return false;
+
+	/*
+	 * Reject if AcquirePlannerLocks would have anything to do.  This is
+	 * simplistic, but there's no need to inquire any more carefully; indeed,
+	 * for current callers it shouldn't even be possible to hit any of these
+	 * checks.
+	 */
+	foreach(lc, plansource->query_list)
+	{
+		Query	   *query = lfirst_node(Query, lc);
+
+		if (query->commandType == CMD_UTILITY)
+			return false;
+		if (query->rtable || query->cteList || query->hasSubLinks)
+			return false;
+	}
+
+	/*
+	 * Reject if AcquireExecutorLocks would have anything to do.  This is
+	 * probably unnecessary given the previous check, but let's be safe.
+	 */
+	foreach(lc, plan->stmt_list)
+	{
+		PlannedStmt *plannedstmt = lfirst_node(PlannedStmt, lc);
+		ListCell   *lc2;
+
+		if (plannedstmt->commandType == CMD_UTILITY)
+			return false;
+
+		/*
+		 * We have to grovel through the rtable because it's likely to contain
+		 * an RTE_RESULT relation, rather than being totally empty.
+		 */
+		foreach(lc2, plannedstmt->rtable)
+		{
+			RangeTblEntry *rte = (RangeTblEntry *) lfirst(lc2);
+
+			if (rte->rtekind == RTE_RELATION)
+				return false;
+		}
+	}
+
+	/*
+	 * Okay, it's simple.  Note that what we've primarily established here is
+	 * that no locks need be taken before checking the plan's is_valid flag.
+	 */
+	return true;
+}
+
+/*
+ * CachedPlanIsSimplyValid: quick check for plan still being valid
+ *
+ * This function must not be used unless CachedPlanAllowsSimpleValidityCheck
+ * previously said it was OK.
+ *
+ * If the plan is valid, and "owner" is not NULL, record a refcount on
+ * the plan in that resowner before returning.  It is caller's responsibility
+ * to be sure that a refcount is held on any plan that's being actively used.
+ *
+ * The code here is unconditionally safe as long as the only use of this
+ * CachedPlanSource is in connection with the particular CachedPlan pointer
+ * that's passed in.  If the plansource were being used for other purposes,
+ * it's possible that its generic plan could be invalidated and regenerated
+ * while the current caller wasn't looking, and then there could be a chance
+ * collision of address between this caller's now-stale plan pointer and the
+ * actual address of the new generic plan.  For current uses, that scenario
+ * can't happen; but with a plansource shared across multiple uses, it'd be
+ * advisable to also save plan->generation and verify that that still matches.
+ */
+bool
+CachedPlanIsSimplyValid(CachedPlanSource *plansource, CachedPlan *plan,
+						ResourceOwner owner)
+{
+	/*
+	 * Careful here: since the caller doesn't necessarily hold a refcount on
+	 * the plan to start with, it's possible that "plan" is a dangling
+	 * pointer.  Don't dereference it until we've verified that it still
+	 * matches the plansource's gplan (which is either valid or NULL).
+	 */
+	Assert(plansource->magic == CACHEDPLANSOURCE_MAGIC);
+
+	/*
+	 * Has cache invalidation fired on this plan?  We can check this right
+	 * away since there are no locks that we'd need to acquire first.
+	 */
+	if (!plansource->is_valid || plan != plansource->gplan || !plan->is_valid)
+		return false;
+
+	Assert(plan->magic == CACHEDPLAN_MAGIC);
+
+	/* Is the search_path still the same as when we made it? */
+	Assert(plansource->search_path != NULL);
+	if (!OverrideSearchPathMatchesCurrent(plansource->search_path))
+		return false;
+
+	/* It's still good.  Bump refcount if requested. */
+	if (owner)
+	{
+		ResourceOwnerEnlargePlanCacheRefs(owner);
+		plan->refcount++;
+		ResourceOwnerRememberPlanCacheRef(owner, plan);
+	}
+
+	return true;
+}
+
+/*
  * CachedPlanSetParentContext: move a CachedPlanSource to a new memory context
  *
  * This can only be applied to unsaved plans; once saved, a plan always
diff --git a/src/backend/utils/resowner/resowner.c b/src/backend/utils/resowner/resowner.c
index 3c39e48..8bc2c4e 100644
--- a/src/backend/utils/resowner/resowner.c
+++ b/src/backend/utils/resowner/resowner.c
@@ -679,6 +679,30 @@ ResourceOwnerReleaseInternal(ResourceOwner owner,
 }
 
 /*
+ * ResourceOwnerReleaseAllPlanCacheRefs
+ *		Release the plancache references (only) held by this owner.
+ *
+ * We might eventually add similar functions for other resource types,
+ * but for now, only this is needed.
+ */
+void
+ResourceOwnerReleaseAllPlanCacheRefs(ResourceOwner owner)
+{
+	ResourceOwner save;
+	Datum		foundres;
+
+	save = CurrentResourceOwner;
+	CurrentResourceOwner = owner;
+	while (ResourceArrayGetAny(&(owner->planrefarr), &foundres))
+	{
+		CachedPlan *res = (CachedPlan *) DatumGetPointer(foundres);
+
+		ReleaseCachedPlan(res, true);
+	}
+	CurrentResourceOwner = save;
+}
+
+/*
  * ResourceOwnerDelete
  *		Delete an owner object and its descendants.
  *
diff --git a/src/include/utils/plancache.h b/src/include/utils/plancache.h
index 6a5953c..037edd0 100644
--- a/src/include/utils/plancache.h
+++ b/src/include/utils/plancache.h
@@ -20,6 +20,8 @@
 #include "nodes/params.h"
 #include "tcop/cmdtag.h"
 #include "utils/queryenvironment.h"
+#include "utils/resowner.h"
+
 
 /* Forward declaration, to avoid including parsenodes.h here */
 struct RawStmt;
@@ -220,6 +222,12 @@ extern CachedPlan *GetCachedPlan(CachedPlanSource *plansource,
 								 QueryEnvironment *queryEnv);
 extern void ReleaseCachedPlan(CachedPlan *plan, bool useResOwner);
 
+extern bool CachedPlanAllowsSimpleValidityCheck(CachedPlanSource *plansource,
+												CachedPlan *plan);
+extern bool CachedPlanIsSimplyValid(CachedPlanSource *plansource,
+									CachedPlan *plan,
+									ResourceOwner owner);
+
 extern CachedExpression *GetCachedExpression(Node *expr);
 extern void FreeCachedExpression(CachedExpression *cexpr);
 
diff --git a/src/include/utils/resowner.h b/src/include/utils/resowner.h
index 850bf69..878f39c 100644
--- a/src/include/utils/resowner.h
+++ b/src/include/utils/resowner.h
@@ -71,6 +71,7 @@ extern void ResourceOwnerRelease(ResourceOwner owner,
 								 ResourceReleasePhase phase,
 								 bool isCommit,
 								 bool isTopLevel);
+extern void ResourceOwnerReleaseAllPlanCacheRefs(ResourceOwner owner);
 extern void ResourceOwnerDelete(ResourceOwner owner);
 extern ResourceOwner ResourceOwnerGetParent(ResourceOwner owner);
 extern void ResourceOwnerNewParent(ResourceOwner owner,
diff --git a/src/pl/plpgsql/src/pl_exec.c b/src/pl/plpgsql/src/pl_exec.c
index d3ad4fa..6f8ae2c 100644
--- a/src/pl/plpgsql/src/pl_exec.c
+++ b/src/pl/plpgsql/src/pl_exec.c
@@ -84,6 +84,13 @@ typedef struct
  * has its own simple-expression EState, which is cleaned up at exit from
  * plpgsql_inline_handler().  DO blocks still use the simple_econtext_stack,
  * though, so that subxact abort cleanup does the right thing.
+ *
+ * (However, if a DO block executes COMMIT or ROLLBACK, then exec_stmt_commit
+ * or exec_stmt_rollback will unlink it from the DO's simple-expression EState
+ * and create a new shared EState that will be used thenceforth.  The original
+ * EState will be cleaned up when we get back to plpgsql_inline_handler.  This
+ * is a bit ugly, but it isn't worth doing better, since scenarios like this
+ * can't result in indefinite accumulation of state trees.)
  */
 typedef struct SimpleEcontextStackEntry
 {
@@ -96,6 +103,15 @@ static EState *shared_simple_eval_estate = NULL;
 static SimpleEcontextStackEntry *simple_econtext_stack = NULL;
 
 /*
+ * In addition to the shared simple-eval EState, we have a shared resource
+ * owner that holds refcounts on the CachedPlans for any "simple" expressions
+ * we have evaluated in the current transaction.  This allows us to avoid
+ * continually grabbing and releasing a plan refcount when a simple expression
+ * is used over and over.
+ */
+static ResourceOwner shared_simple_eval_resowner = NULL;
+
+/*
  * Memory management within a plpgsql function generally works with three
  * contexts:
  *
@@ -314,7 +330,8 @@ static int	exec_stmt_set(PLpgSQL_execstate *estate,
 static void plpgsql_estate_setup(PLpgSQL_execstate *estate,
 								 PLpgSQL_function *func,
 								 ReturnSetInfo *rsi,
-								 EState *simple_eval_estate);
+								 EState *simple_eval_estate,
+								 ResourceOwner simple_eval_resowner);
 static void exec_eval_cleanup(PLpgSQL_execstate *estate);
 
 static void exec_prepare_plan(PLpgSQL_execstate *estate,
@@ -440,16 +457,19 @@ static char *format_preparedparamsdata(PLpgSQL_execstate *estate,
  *
  * This is also used to execute inline code blocks (DO blocks).  The only
  * difference that this code is aware of is that for a DO block, we want
- * to use a private simple_eval_estate, which is created and passed in by
- * the caller.  For regular functions, pass NULL, which implies using
- * shared_simple_eval_estate.  (When using a private simple_eval_estate,
+ * to use a private simple_eval_estate and a private simple_eval_resowner,
+ * which are created and passed in by the caller.  For regular functions,
+ * pass NULL, which implies using shared_simple_eval_estate and
+ * shared_simple_eval_resowner.  (When using a private simple_eval_estate,
  * we must also use a private cast hashtable, but that's taken care of
  * within plpgsql_estate_setup.)
  * ----------
  */
 Datum
 plpgsql_exec_function(PLpgSQL_function *func, FunctionCallInfo fcinfo,
-					  EState *simple_eval_estate, bool atomic)
+					  EState *simple_eval_estate,
+					  ResourceOwner simple_eval_resowner,
+					  bool atomic)
 {
 	PLpgSQL_execstate estate;
 	ErrorContextCallback plerrcontext;
@@ -460,7 +480,7 @@ plpgsql_exec_function(PLpgSQL_function *func, FunctionCallInfo fcinfo,
 	 * Setup the execution state
 	 */
 	plpgsql_estate_setup(&estate, func, (ReturnSetInfo *) fcinfo->resultinfo,
-						 simple_eval_estate);
+						 simple_eval_estate, simple_eval_resowner);
 	estate.atomic = atomic;
 
 	/*
@@ -897,7 +917,7 @@ plpgsql_exec_trigger(PLpgSQL_function *func,
 	/*
 	 * Setup the execution state
 	 */
-	plpgsql_estate_setup(&estate, func, NULL, NULL);
+	plpgsql_estate_setup(&estate, func, NULL, NULL, NULL);
 	estate.trigdata = trigdata;
 
 	/*
@@ -1135,7 +1155,7 @@ plpgsql_exec_event_trigger(PLpgSQL_function *func, EventTriggerData *trigdata)
 	/*
 	 * Setup the execution state
 	 */
-	plpgsql_estate_setup(&estate, func, NULL, NULL);
+	plpgsql_estate_setup(&estate, func, NULL, NULL, NULL);
 	estate.evtrigdata = trigdata;
 
 	/*
@@ -2319,6 +2339,7 @@ exec_stmt_call(PLpgSQL_execstate *estate, PLpgSQL_stmt_call *stmt)
 		 * some internal state.
 		 */
 		estate->simple_eval_estate = NULL;
+		estate->simple_eval_resowner = NULL;
 		plpgsql_create_econtext(estate);
 	}
 
@@ -3874,7 +3895,8 @@ static void
 plpgsql_estate_setup(PLpgSQL_execstate *estate,
 					 PLpgSQL_function *func,
 					 ReturnSetInfo *rsi,
-					 EState *simple_eval_estate)
+					 EState *simple_eval_estate,
+					 ResourceOwner simple_eval_resowner)
 {
 	HASHCTL		ctl;
 
@@ -3965,6 +3987,11 @@ plpgsql_estate_setup(PLpgSQL_execstate *estate,
 		estate->cast_hash = shared_cast_hash;
 		estate->cast_hash_context = shared_cast_context;
 	}
+	/* likewise for the simple-expression resource owner */
+	if (simple_eval_resowner)
+		estate->simple_eval_resowner = simple_eval_resowner;
+	else
+		estate->simple_eval_resowner = shared_simple_eval_resowner;
 
 	/*
 	 * We start with no stmt_mcontext; one will be created only if needed.
@@ -4825,6 +4852,7 @@ exec_stmt_commit(PLpgSQL_execstate *estate, PLpgSQL_stmt_commit *stmt)
 	}
 
 	estate->simple_eval_estate = NULL;
+	estate->simple_eval_resowner = NULL;
 	plpgsql_create_econtext(estate);
 
 	return PLPGSQL_RC_OK;
@@ -4847,6 +4875,7 @@ exec_stmt_rollback(PLpgSQL_execstate *estate, PLpgSQL_stmt_rollback *stmt)
 	}
 
 	estate->simple_eval_estate = NULL;
+	estate->simple_eval_resowner = NULL;
 	plpgsql_create_econtext(estate);
 
 	return PLPGSQL_RC_OK;
@@ -6059,8 +6088,6 @@ loop_exit:
  * someone might redefine a SQL function that had been inlined into the simple
  * expression.  That cannot cause a simple expression to become non-simple (or
  * vice versa), but we do have to handle replacing the expression tree.
- * Fortunately it's normally inexpensive to call SPI_plan_get_cached_plan for
- * a simple expression.
  *
  * Note: if pass-by-reference, the result is in the eval_mcontext.
  * It will be freed when exec_eval_cleanup is done.
@@ -6076,7 +6103,7 @@ exec_eval_simple_expr(PLpgSQL_execstate *estate,
 {
 	ExprContext *econtext = estate->eval_econtext;
 	LocalTransactionId curlxid = MyProc->lxid;
-	CachedPlan *cplan;
+	ParamListInfo paramLI;
 	void	   *save_setup_arg;
 	bool		need_snapshot;
 	MemoryContext oldcontext;
@@ -6090,29 +6117,92 @@ exec_eval_simple_expr(PLpgSQL_execstate *estate,
 	/*
 	 * If expression is in use in current xact, don't touch it.
 	 */
-	if (expr->expr_simple_in_use && expr->expr_simple_lxid == curlxid)
+	if (unlikely(expr->expr_simple_in_use) &&
+		expr->expr_simple_lxid == curlxid)
 		return false;
 
 	/*
-	 * Revalidate cached plan, so that we will notice if it became stale. (We
-	 * need to hold a refcount while using the plan, anyway.)  If replanning
-	 * is needed, do that work in the eval_mcontext.
+	 * Check to see if the cached plan has been invalidated.  If not, and this
+	 * is the first use in the current transaction, save a plan refcount in
+	 * the simple-expression resowner.
 	 */
-	oldcontext = MemoryContextSwitchTo(get_eval_mcontext(estate));
-	cplan = SPI_plan_get_cached_plan(expr->plan);
-	MemoryContextSwitchTo(oldcontext);
+	if (likely(CachedPlanIsSimplyValid(expr->expr_simple_plansource,
+									   expr->expr_simple_plan,
+									   (expr->expr_simple_plan_lxid != curlxid ?
+										estate->simple_eval_resowner : NULL))))
+	{
+		/*
+		 * It's still good, so just remember that we have a refcount on the
+		 * plan in the current transaction.  (If we already had one, this
+		 * assignment is a no-op.)
+		 */
+		expr->expr_simple_plan_lxid = curlxid;
+	}
+	else
+	{
+		/* Need to replan */
+		CachedPlan *cplan;
 
-	/*
-	 * We can't get a failure here, because the number of CachedPlanSources in
-	 * the SPI plan can't change from what exec_simple_check_plan saw; it's a
-	 * property of the raw parsetree generated from the query text.
-	 */
-	Assert(cplan != NULL);
+		/*
+		 * If we have a valid refcount on some previous version of the plan,
+		 * release it, so we don't leak plans intra-transaction.
+		 */
+		if (expr->expr_simple_plan_lxid == curlxid)
+		{
+			ResourceOwner saveResourceOwner = CurrentResourceOwner;
 
-	/* If it got replanned, update our copy of the simple expression */
-	if (cplan->generation != expr->expr_simple_generation)
-	{
+			CurrentResourceOwner = estate->simple_eval_resowner;
+			ReleaseCachedPlan(expr->expr_simple_plan, true);
+			CurrentResourceOwner = saveResourceOwner;
+			expr->expr_simple_plan = NULL;
+			expr->expr_simple_plan_lxid = InvalidLocalTransactionId;
+		}
+
+		/* Do the replanning work in the eval_mcontext */
+		oldcontext = MemoryContextSwitchTo(get_eval_mcontext(estate));
+		cplan = SPI_plan_get_cached_plan(expr->plan);
+		MemoryContextSwitchTo(oldcontext);
+
+		/*
+		 * We can't get a failure here, because the number of
+		 * CachedPlanSources in the SPI plan can't change from what
+		 * exec_simple_check_plan saw; it's a property of the raw parsetree
+		 * generated from the query text.
+		 */
+		Assert(cplan != NULL);
+
+		/*
+		 * These tests probably can't fail either, but if they do, cope by
+		 * declaring the plan to be non-simple.  On success, we'll acquire a
+		 * refcount on the new plan, stored in simple_eval_resowner.
+		 */
+		if (CachedPlanAllowsSimpleValidityCheck(expr->expr_simple_plansource,
+												cplan) &&
+			CachedPlanIsSimplyValid(expr->expr_simple_plansource, cplan,
+									estate->simple_eval_resowner))
+		{
+			/* Remember that we have the refcount */
+			expr->expr_simple_plan = cplan;
+			expr->expr_simple_plan_lxid = curlxid;
+		}
+		else
+		{
+			/* Release SPI_plan_get_cached_plan's refcount */
+			ReleaseCachedPlan(cplan, true);
+			/* Mark expression as non-simple, and fail */
+			expr->expr_simple_expr = NULL;
+			return false;
+		}
+
+		/*
+		 * SPI_plan_get_cached_plan acquired a plan refcount stored in the
+		 * active resowner.  We don't need that anymore, so release it.
+		 */
+		ReleaseCachedPlan(cplan, true);
+
+		/* Extract desired scalar expression from cached plan */
 		exec_save_simple_expr(expr, cplan);
+
 		/* better recheck r/w safety, as it could change due to inlining */
 		if (expr->rwparam >= 0)
 			exec_check_rw_parameter(expr, expr->rwparam);
@@ -6128,16 +6218,24 @@ exec_eval_simple_expr(PLpgSQL_execstate *estate,
 	 * Set up ParamListInfo to pass to executor.  For safety, save and restore
 	 * estate->paramLI->parserSetupArg around our use of the param list.
 	 */
-	save_setup_arg = estate->paramLI->parserSetupArg;
+	paramLI = estate->paramLI;
+	save_setup_arg = paramLI->parserSetupArg;
 
-	econtext->ecxt_param_list_info = setup_param_list(estate, expr);
+	/*
+	 * We can skip using setup_param_list() in favor of just doing this
+	 * unconditionally, because there's no need for the optimization of
+	 * possibly setting ecxt_param_list_info to NULL; we've already forced use
+	 * of a generic plan.
+	 */
+	paramLI->parserSetupArg = (void *) expr;
+	econtext->ecxt_param_list_info = paramLI;
 
 	/*
 	 * Prepare the expression for execution, if it's not been done already in
 	 * the current transaction.  (This will be forced to happen if we called
 	 * exec_save_simple_expr above.)
 	 */
-	if (expr->expr_simple_lxid != curlxid)
+	if (unlikely(expr->expr_simple_lxid != curlxid))
 	{
 		oldcontext = MemoryContextSwitchTo(estate->simple_eval_estate->es_query_cxt);
 		expr->expr_simple_state =
@@ -6185,7 +6283,7 @@ exec_eval_simple_expr(PLpgSQL_execstate *estate,
 
 	econtext->ecxt_param_list_info = NULL;
 
-	estate->paramLI->parserSetupArg = save_setup_arg;
+	paramLI->parserSetupArg = save_setup_arg;
 
 	if (need_snapshot)
 		PopActiveSnapshot();
@@ -6193,11 +6291,6 @@ exec_eval_simple_expr(PLpgSQL_execstate *estate,
 	MemoryContextSwitchTo(oldcontext);
 
 	/*
-	 * Now we can release our refcount on the cached plan.
-	 */
-	ReleaseCachedPlan(cplan, true);
-
-	/*
 	 * That's it.
 	 */
 	return true;
@@ -7984,10 +8077,35 @@ exec_simple_check_plan(PLpgSQL_execstate *estate, PLpgSQL_expr *expr)
 	/* Can't fail, because we checked for a single CachedPlanSource above */
 	Assert(cplan != NULL);
 
-	/* Share the remaining work with replan code path */
-	exec_save_simple_expr(expr, cplan);
+	/*
+	 * Verify that plancache.c thinks the plan is simple enough to use
+	 * CachedPlanIsSimplyValid.  Given the restrictions above, it's unlikely
+	 * that this could fail, but if it does, just treat plan as not simple.
+	 */
+	if (CachedPlanAllowsSimpleValidityCheck(plansource, cplan))
+	{
+		/*
+		 * OK, use CachedPlanIsSimplyValid to save a refcount on the plan in
+		 * the simple-expression resowner.  This shouldn't fail either, but if
+		 * somehow it does, again we can cope by treating plan as not simple.
+		 */
+		if (CachedPlanIsSimplyValid(plansource, cplan,
+									estate->simple_eval_resowner))
+		{
+			/* Remember that we have the refcount */
+			expr->expr_simple_plansource = plansource;
+			expr->expr_simple_plan = cplan;
+			expr->expr_simple_plan_lxid = MyProc->lxid;
+
+			/* Share the remaining work with the replan code path */
+			exec_save_simple_expr(expr, cplan);
+		}
+	}
 
-	/* Release our plan refcount */
+	/*
+	 * Release the plan refcount obtained by SPI_plan_get_cached_plan.  (This
+	 * refcount is held by the wrong resowner, so we can't just repurpose it.)
+	 */
 	ReleaseCachedPlan(cplan, true);
 }
 
@@ -8060,7 +8178,6 @@ exec_save_simple_expr(PLpgSQL_expr *expr, CachedPlan *cplan)
 	 * current transaction".
 	 */
 	expr->expr_simple_expr = tle_expr;
-	expr->expr_simple_generation = cplan->generation;
 	expr->expr_simple_state = NULL;
 	expr->expr_simple_in_use = false;
 	expr->expr_simple_lxid = InvalidLocalTransactionId;
@@ -8196,7 +8313,7 @@ exec_set_found(PLpgSQL_execstate *estate, bool state)
  *
  * We may need to create a new shared_simple_eval_estate too, if there's not
  * one already for the current transaction.  The EState will be cleaned up at
- * transaction end.
+ * transaction end.  Ditto for shared_simple_eval_resowner.
  */
 static void
 plpgsql_create_econtext(PLpgSQL_execstate *estate)
@@ -8208,8 +8325,11 @@ plpgsql_create_econtext(PLpgSQL_execstate *estate)
 	 * one already in the current transaction.  The EState is made a child of
 	 * TopTransactionContext so it will have the right lifespan.
 	 *
-	 * Note that this path is never taken when executing a DO block; the
-	 * required EState was already made by plpgsql_inline_handler.
+	 * Note that this path is never taken when beginning a DO block; the
+	 * required EState was already made by plpgsql_inline_handler.  However,
+	 * if the DO block executes COMMIT or ROLLBACK, then we'll come here and
+	 * make a shared EState to use for the rest of the DO block.  That's OK;
+	 * see the comments above struct SimpleEcontextStackEntry.
 	 */
 	if (estate->simple_eval_estate == NULL)
 	{
@@ -8225,6 +8345,18 @@ plpgsql_create_econtext(PLpgSQL_execstate *estate)
 	}
 
 	/*
+	 * Likewise for the simple-expression resource owner.
+	 */
+	if (estate->simple_eval_resowner == NULL)
+	{
+		if (shared_simple_eval_resowner == NULL)
+			shared_simple_eval_resowner =
+				ResourceOwnerCreate(TopTransactionResourceOwner,
+									"PL/pgSQL simple expressions");
+		estate->simple_eval_resowner = shared_simple_eval_resowner;
+	}
+
+	/*
 	 * Create a child econtext for the current function.
 	 */
 	estate->eval_econtext = CreateExprContext(estate->simple_eval_estate);
@@ -8270,29 +8402,40 @@ plpgsql_destroy_econtext(PLpgSQL_execstate *estate)
  * plpgsql_xact_cb --- post-transaction-commit-or-abort cleanup
  *
  * If a simple-expression EState was created in the current transaction,
- * it has to be cleaned up.
+ * it has to be cleaned up.  The same for the simple-expression resowner.
  */
 void
 plpgsql_xact_cb(XactEvent event, void *arg)
 {
 	/*
-	 * If we are doing a clean transaction shutdown, free the EState (so that
-	 * any remaining resources will be released correctly). In an abort, we
-	 * expect the regular abort recovery procedures to release everything of
-	 * interest.
+	 * If we are doing a clean transaction shutdown, free the EState and tell
+	 * the resowner to release whatever plancache references it has, so that
+	 * all remaining resources will be released correctly.  (We don't need to
+	 * actually delete the resowner here; deletion of the
+	 * TopTransactionResourceOwner will take care of that.)
+	 *
+	 * In an abort, we expect the regular abort recovery procedures to release
+	 * everything of interest, so just clear our pointers.
 	 */
-	if (event == XACT_EVENT_COMMIT || event == XACT_EVENT_PREPARE)
+	if (event == XACT_EVENT_COMMIT ||
+		event == XACT_EVENT_PARALLEL_COMMIT ||
+		event == XACT_EVENT_PREPARE)
 	{
 		simple_econtext_stack = NULL;
 
 		if (shared_simple_eval_estate)
 			FreeExecutorState(shared_simple_eval_estate);
 		shared_simple_eval_estate = NULL;
+		if (shared_simple_eval_resowner)
+			ResourceOwnerReleaseAllPlanCacheRefs(shared_simple_eval_resowner);
+		shared_simple_eval_resowner = NULL;
 	}
-	else if (event == XACT_EVENT_ABORT)
+	else if (event == XACT_EVENT_ABORT ||
+			 event == XACT_EVENT_PARALLEL_ABORT)
 	{
 		simple_econtext_stack = NULL;
 		shared_simple_eval_estate = NULL;
+		shared_simple_eval_resowner = NULL;
 	}
 }
 
diff --git a/src/pl/plpgsql/src/pl_handler.c b/src/pl/plpgsql/src/pl_handler.c
index b434818..97cf849 100644
--- a/src/pl/plpgsql/src/pl_handler.c
+++ b/src/pl/plpgsql/src/pl_handler.c
@@ -262,7 +262,9 @@ plpgsql_call_handler(PG_FUNCTION_ARGS)
 			retval = (Datum) 0;
 		}
 		else
-			retval = plpgsql_exec_function(func, fcinfo, NULL, !nonatomic);
+			retval = plpgsql_exec_function(func, fcinfo,
+										   NULL, NULL,
+										   !nonatomic);
 	}
 	PG_FINALLY();
 	{
@@ -297,6 +299,7 @@ plpgsql_inline_handler(PG_FUNCTION_ARGS)
 	PLpgSQL_function *func;
 	FmgrInfo	flinfo;
 	EState	   *simple_eval_estate;
+	ResourceOwner simple_eval_resowner;
 	Datum		retval;
 	int			rc;
 
@@ -323,22 +326,33 @@ plpgsql_inline_handler(PG_FUNCTION_ARGS)
 	flinfo.fn_oid = InvalidOid;
 	flinfo.fn_mcxt = CurrentMemoryContext;
 
-	/* Create a private EState for simple-expression execution */
+	/*
+	 * Create a private EState and resowner for simple-expression execution.
+	 * Notice that these are NOT tied to transaction-level resources; they
+	 * must survive any COMMIT/ROLLBACK the DO block executes, since we will
+	 * unconditionally try to clean them up below.  (Hence, be wary of adding
+	 * anything that could fail between here and the PG_TRY block.)
+	 */
 	simple_eval_estate = CreateExecutorState();
+	simple_eval_resowner =
+		ResourceOwnerCreate(NULL, "PL/pgSQL DO block simple expressions");
 
 	/* And run the function */
 	PG_TRY();
 	{
-		retval = plpgsql_exec_function(func, fake_fcinfo, simple_eval_estate, codeblock->atomic);
+		retval = plpgsql_exec_function(func, fake_fcinfo,
+									   simple_eval_estate,
+									   simple_eval_resowner,
+									   codeblock->atomic);
 	}
 	PG_CATCH();
 	{
 		/*
 		 * We need to clean up what would otherwise be long-lived resources
 		 * accumulated by the failed DO block, principally cached plans for
-		 * statements (which can be flushed with plpgsql_free_function_memory)
-		 * and execution trees for simple expressions, which are in the
-		 * private EState.
+		 * statements (which can be flushed by plpgsql_free_function_memory),
+		 * execution trees for simple expressions, which are in the private
+		 * EState, and cached-plan refcounts held by the private resowner.
 		 *
 		 * Before releasing the private EState, we must clean up any
 		 * simple_econtext_stack entries pointing into it, which we can do by
@@ -351,8 +365,10 @@ plpgsql_inline_handler(PG_FUNCTION_ARGS)
 						   GetCurrentSubTransactionId(),
 						   0, NULL);
 
-		/* Clean up the private EState */
+		/* Clean up the private EState and resowner */
 		FreeExecutorState(simple_eval_estate);
+		ResourceOwnerReleaseAllPlanCacheRefs(simple_eval_resowner);
+		ResourceOwnerDelete(simple_eval_resowner);
 
 		/* Function should now have no remaining use-counts ... */
 		func->use_count--;
@@ -366,8 +382,10 @@ plpgsql_inline_handler(PG_FUNCTION_ARGS)
 	}
 	PG_END_TRY();
 
-	/* Clean up the private EState */
+	/* Clean up the private EState and resowner */
 	FreeExecutorState(simple_eval_estate);
+	ResourceOwnerReleaseAllPlanCacheRefs(simple_eval_resowner);
+	ResourceOwnerDelete(simple_eval_resowner);
 
 	/* Function should now have no remaining use-counts ... */
 	func->use_count--;
diff --git a/src/pl/plpgsql/src/plpgsql.h b/src/pl/plpgsql/src/plpgsql.h
index 69df330..1af2595 100644
--- a/src/pl/plpgsql/src/plpgsql.h
+++ b/src/pl/plpgsql/src/plpgsql.h
@@ -231,12 +231,21 @@ typedef struct PLpgSQL_expr
 
 	/* fields for "simple expression" fast-path execution: */
 	Expr	   *expr_simple_expr;	/* NULL means not a simple expr */
-	int			expr_simple_generation; /* plancache generation we checked */
 	Oid			expr_simple_type;	/* result type Oid, if simple */
 	int32		expr_simple_typmod; /* result typmod, if simple */
 	bool		expr_simple_mutable;	/* true if simple expr is mutable */
 
 	/*
+	 * If the expression was ever determined to be simple, we remember its
+	 * CachedPlanSource and CachedPlan here.  If expr_simple_plan_lxid matches
+	 * current LXID, then we hold a refcount on expr_simple_plan in the
+	 * current transaction.  Otherwise we need to get one before re-using it.
+	 */
+	CachedPlanSource *expr_simple_plansource;	/* extracted from "plan" */
+	CachedPlan *expr_simple_plan;	/* extracted from "plan" */
+	LocalTransactionId expr_simple_plan_lxid;
+
+	/*
 	 * if expr is simple AND prepared in current transaction,
 	 * expr_simple_state and expr_simple_in_use are valid. Test validity by
 	 * seeing if expr_simple_lxid matches current LXID.  (If not,
@@ -1082,8 +1091,9 @@ typedef struct PLpgSQL_execstate
 	 */
 	ParamListInfo paramLI;
 
-	/* EState to use for "simple" expression evaluation */
+	/* EState and resowner to use for "simple" expression evaluation */
 	EState	   *simple_eval_estate;
+	ResourceOwner simple_eval_resowner;
 
 	/* lookup table to use for executing type casts */
 	HTAB	   *cast_hash;
@@ -1268,6 +1278,7 @@ extern void _PG_init(void);
 extern Datum plpgsql_exec_function(PLpgSQL_function *func,
 								   FunctionCallInfo fcinfo,
 								   EState *simple_eval_estate,
+								   ResourceOwner simple_eval_resowner,
 								   bool atomic);
 extern HeapTuple plpgsql_exec_trigger(PLpgSQL_function *func,
 									  TriggerData *trigdata);

diff --git a/src/backend/catalog/namespace.c b/src/backend/catalog/namespace.c
index 5ff7824..2ec2301 100644
--- a/src/backend/catalog/namespace.c
+++ b/src/backend/catalog/namespace.c
@@ -126,6 +126,11 @@
  * namespaceUser is the userid the path has been computed for.
  *
  * Note: all data pointed to by these List variables is in TopMemoryContext.
+ *
+ * activePathGeneration is incremented whenever the effective values of
+ * activeSearchPath/activeCreationNamespace/activeTempCreationPending change.
+ * This can be used to quickly detect whether any change has happened since
+ * a previous examination of the search path state.
  */
 
 /* These variables define the actually active state: */
@@ -138,6 +143,9 @@ static Oid	activeCreationNamespace = InvalidOid;
 /* if true, activeCreationNamespace is wrong, it should be temp namespace */
 static bool activeTempCreationPending = false;
 
+/* current generation counter; make sure this is never zero */
+static uint64 activePathGeneration = 1;
+
 /* These variables are the values last derived from namespace_search_path: */
 
 static List *baseSearchPath = NIL;
@@ -3373,6 +3381,7 @@ GetOverrideSearchPath(MemoryContext context)
 		schemas = list_delete_first(schemas);
 	}
 	result->schemas = schemas;
+	result->generation = activePathGeneration;
 
 	MemoryContextSwitchTo(oldcxt);
 
@@ -3393,12 +3402,18 @@ CopyOverrideSearchPath(OverrideSearchPath *path)
 	result->schemas = list_copy(path->schemas);
 	result->addCatalog = path->addCatalog;
 	result->addTemp = path->addTemp;
+	result->generation = path->generation;
 
 	return result;
 }
 
 /*
  * OverrideSearchPathMatchesCurrent - does path match current setting?
+ *
+ * This is tested over and over in some common code paths, and in the typical
+ * scenario where the active search path seldom changes, it'll always succeed.
+ * We make that case fast by keeping a generation counter that is advanced
+ * whenever the active search path changes.
  */
 bool
 OverrideSearchPathMatchesCurrent(OverrideSearchPath *path)
@@ -3408,6 +3423,10 @@ OverrideSearchPathMatchesCurrent(OverrideSearchPath *path)
 
 	recomputeNamespacePath();
 
+	/* Quick out if already known equal to active path. */
+	if (path->generation == activePathGeneration)
+		return true;
+
 	/* We scan down the activeSearchPath to see if it matches the input. */
 	lc = list_head(activeSearchPath);
 
@@ -3440,6 +3459,13 @@ OverrideSearchPathMatchesCurrent(OverrideSearchPath *path)
 	}
 	if (lc)
 		return false;
+
+	/*
+	 * Update path->generation so that future tests will return quickly, so
+	 * long as the active search path doesn't change.
+	 */
+	path->generation = activePathGeneration;
+
 	return true;
 }
 
@@ -3510,6 +3536,14 @@ PushOverrideSearchPath(OverrideSearchPath *newpath)
 	activeCreationNamespace = entry->creationNamespace;
 	activeTempCreationPending = false;	/* XXX is this OK? */
 
+	/*
+	 * We always increment activePathGeneration when pushing/popping an
+	 * override path.  In current usage, these actions always change the
+	 * effective path state, so there's no value in checking to see if it
+	 * didn't change.
+	 */
+	activePathGeneration++;
+
 	MemoryContextSwitchTo(oldcxt);
 }
 
@@ -3551,6 +3585,9 @@ PopOverrideSearchPath(void)
 		activeCreationNamespace = baseCreationNamespace;
 		activeTempCreationPending = baseTempCreationPending;
 	}
+
+	/* As above, the generation always increments. */
+	activePathGeneration++;
 }
 
 
@@ -3707,6 +3744,7 @@ recomputeNamespacePath(void)
 	ListCell   *l;
 	bool		temp_missing;
 	Oid			firstNS;
+	bool		pathChanged;
 	MemoryContext oldcxt;
 
 	/* Do nothing if an override search spec is active. */
@@ -3814,18 +3852,31 @@ recomputeNamespacePath(void)
 		oidlist = lcons_oid(myTempNamespace, oidlist);
 
 	/*
-	 * Now that we've successfully built the new list of namespace OIDs, save
-	 * it in permanent storage.
+	 * We want to detect the case where the effective value of the base search
+	 * path variables didn't change.  As long as we're doing so, we can avoid
+	 * copying the OID list unncessarily.
 	 */
-	oldcxt = MemoryContextSwitchTo(TopMemoryContext);
-	newpath = list_copy(oidlist);
-	MemoryContextSwitchTo(oldcxt);
+	if (baseCreationNamespace == firstNS &&
+		baseTempCreationPending == temp_missing &&
+		equal(oidlist, baseSearchPath))
+	{
+		pathChanged = false;
+	}
+	else
+	{
+		pathChanged = true;
+
+		/* Must save OID list in permanent storage. */
+		oldcxt = MemoryContextSwitchTo(TopMemoryContext);
+		newpath = list_copy(oidlist);
+		MemoryContextSwitchTo(oldcxt);
 
-	/* Now safe to assign to state variables. */
-	list_free(baseSearchPath);
-	baseSearchPath = newpath;
-	baseCreationNamespace = firstNS;
-	baseTempCreationPending = temp_missing;
+		/* Now safe to assign to state variables. */
+		list_free(baseSearchPath);
+		baseSearchPath = newpath;
+		baseCreationNamespace = firstNS;
+		baseTempCreationPending = temp_missing;
+	}
 
 	/* Mark the path valid. */
 	baseSearchPathValid = true;
@@ -3836,6 +3887,16 @@ recomputeNamespacePath(void)
 	activeCreationNamespace = baseCreationNamespace;
 	activeTempCreationPending = baseTempCreationPending;
 
+	/*
+	 * Bump the generation only if something actually changed.  (Notice that
+	 * what we compared to was the old state of the base path variables; so
+	 * this does not deal with the situation where we have just popped an
+	 * override path and restored the prior state of the base path.  Instead
+	 * we rely on the override-popping logic to have bumped the generation.)
+	 */
+	if (pathChanged)
+		activePathGeneration++;
+
 	/* Clean up. */
 	pfree(rawname);
 	list_free(namelist);
@@ -4054,6 +4115,8 @@ AtEOXact_Namespace(bool isCommit, bool parallel)
 		activeSearchPath = baseSearchPath;
 		activeCreationNamespace = baseCreationNamespace;
 		activeTempCreationPending = baseTempCreationPending;
+		/* Always bump generation --- see note in recomputeNamespacePath */
+		activePathGeneration++;
 	}
 }
 
@@ -4109,6 +4172,8 @@ AtEOSubXact_Namespace(bool isCommit, SubTransactionId mySubid,
 		overrideStack = list_delete_first(overrideStack);
 		list_free(entry->searchPath);
 		pfree(entry);
+		/* Always bump generation --- see note in recomputeNamespacePath */
+		activePathGeneration++;
 	}
 
 	/* Activate the next level down. */
@@ -4118,6 +4183,12 @@ AtEOSubXact_Namespace(bool isCommit, SubTransactionId mySubid,
 		activeSearchPath = entry->searchPath;
 		activeCreationNamespace = entry->creationNamespace;
 		activeTempCreationPending = false;	/* XXX is this OK? */
+
+		/*
+		 * It's probably unnecessary to bump generation here, but this should
+		 * not be a performance-critical case, so better to be over-cautious.
+		 */
+		activePathGeneration++;
 	}
 	else
 	{
@@ -4125,6 +4196,12 @@ AtEOSubXact_Namespace(bool isCommit, SubTransactionId mySubid,
 		activeSearchPath = baseSearchPath;
 		activeCreationNamespace = baseCreationNamespace;
 		activeTempCreationPending = baseTempCreationPending;
+
+		/*
+		 * If we popped an override stack entry, then we already bumped the
+		 * generation above.  If we did not, then the above assignments did
+		 * nothing and we need not bump the generation.
+		 */
 	}
 }
 
@@ -4264,6 +4341,7 @@ InitializeSearchPath(void)
 		activeSearchPath = baseSearchPath;
 		activeCreationNamespace = baseCreationNamespace;
 		activeTempCreationPending = baseTempCreationPending;
+		activePathGeneration++; /* pro forma */
 	}
 	else
 	{
diff --git a/src/include/catalog/namespace.h b/src/include/catalog/namespace.h
index 3e3a192..2456c08 100644
--- a/src/include/catalog/namespace.h
+++ b/src/include/catalog/namespace.h
@@ -49,12 +49,17 @@ typedef enum TempNamespaceStatus
 
 /*
  *	Structure for xxxOverrideSearchPath functions
+ *
+ * The generation counter is private to namespace.c and shouldn't be touched
+ * by other code.  It can be initialized to zero if necessary (that means
+ * "not known equal to the current active path").
  */
 typedef struct OverrideSearchPath
 {
 	List	   *schemas;		/* OIDs of explicitly named schemas */
 	bool		addCatalog;		/* implicitly prepend pg_catalog? */
 	bool		addTemp;		/* implicitly prepend temp schema? */
+	uint64		generation;		/* for quick detection of equality to active */
 } OverrideSearchPath;
 
 /*