Improve performance of "simple expressions" in PL/pgSQL.

For relatively simple expressions (say, "x + 1" or "x > 0"), plpgsql's
management overhead exceeds the cost of evaluating the expression.
This patch substantially improves that situation, providing roughly
2X speedup for such trivial expressions.

First, add infrastructure in the plancache to allow fast re-validation
of cached plans that contain no table access, and hence need no locks.
Teach plpgsql to use this infrastructure for expressions that it's
already deemed "simple" (which in particular will never contain table
references).

The fast path still requires checking that search_path hasn't changed,
so provide a fast path for OverrideSearchPathMatchesCurrent by
counting changes that have occurred to the active search path in the
current session.  This is simplistic but seems enough for now, seeing
that PushOverrideSearchPath is not used in any performance-critical
cases.

Second, manage the refcounts on simple expressions' cached plans using
a transaction-lifespan resource owner, so that we only need to take
and release an expression's refcount once per transaction not once per
expression evaluation.  The management of this resource owner exactly
parallels the existing management of plpgsql's simple-expression EState.

Add some regression tests covering this area, in particular verifying
that expression caching doesn't break semantics for search_path changes.

Patch by me, but it owes something to previous work by Amit Langote,
who recognized that getting rid of plancache-related overhead would
be a useful thing to do here.  Also thanks to Andres Freund for review.

Discussion: https://postgr.es/m/CAFj8pRDRVfLdAxsWeVLzCAbkLFZhW549K+67tpOc-faC8uH8zw@mail.gmail.com

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
 	{

src/backend/utils/cache/plancache.c

@@ -1277,6 +1277,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
  *

src/backend/utils/resowner/resowner.c

@@ -678,6 +678,30 @@ ResourceOwnerReleaseInternal(ResourceOwner owner,
 	CurrentResourceOwner = save;
 }
 
+/*
+ * 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.

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;
 
 /*

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);
 

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,

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

src/pl/plpgsql/src/expected/plpgsql_simple.out

+--
+-- 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)
+

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;
 
@@ -324,28 +327,33 @@ plpgsql_inline_handler(PG_FUNCTION_ARGS)
 	flinfo.fn_mcxt = CurrentMemoryContext;
 
 	/*
-	 * Create a private EState for simple-expression execution.  Notice that
-	 * this is NOT tied to transaction-level resources; it must survive any
-	 * COMMIT/ROLLBACK the DO block executes, since we will unconditionally
-	 * try to clean it up below.  (Hence, be wary of adding anything that
-	 * could fail between here and the PG_TRY block.)  See the comments for
-	 * shared_simple_eval_estate.
+	 * 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.)  See the
+	 * comments for shared_simple_eval_estate.
 	 */
 	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
@@ -358,8 +366,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--;
@@ -373,8 +383,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--;

src/pl/plpgsql/src/plpgsql.h

@@ -231,11 +231,20 @@ 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
@@ -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);

src/pl/plpgsql/src/sql/plpgsql_simple.sql

+--
+-- 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();