13 files changed, 179 insertions, 20 deletions

diff --git a/src/backend/executor/execParallel.c b/src/backend/executor/execParallel.c
index c30b3485dd5..29e450a571c 100644
--- a/src/backend/executor/execParallel.c
+++ b/src/backend/executor/execParallel.c
@@ -143,6 +143,7 @@ ExecSerializePlan(Plan *plan, EState *estate)
 	pstmt->relationOids = NIL;
 	pstmt->invalItems = NIL;	/* workers can't replan anyway... */
 	pstmt->hasRowSecurity = false;
+	pstmt->hasForeignJoin = false;
 
 	/* Return serialized copy of our dummy PlannedStmt. */
 	return nodeToString(pstmt);
diff --git a/src/backend/foreign/foreign.c b/src/backend/foreign/foreign.c
index c24b11b685c..47c00af74f9 100644
--- a/src/backend/foreign/foreign.c
+++ b/src/backend/foreign/foreign.c
@@ -31,6 +31,7 @@
 extern Datum pg_options_to_table(PG_FUNCTION_ARGS);
 extern Datum postgresql_fdw_validator(PG_FUNCTION_ARGS);
 
+static HeapTuple find_user_mapping(Oid userid, Oid serverid);
 
 /*
  * GetForeignDataWrapper -	look up the foreign-data wrapper by OID.
@@ -174,23 +175,7 @@ GetUserMapping(Oid userid, Oid serverid)
 	bool		isnull;
 	UserMapping *um;
 
-	tp = SearchSysCache2(USERMAPPINGUSERSERVER,
-						 ObjectIdGetDatum(userid),
-						 ObjectIdGetDatum(serverid));
-
-	if (!HeapTupleIsValid(tp))
-	{
-		/* Not found for the specific user -- try PUBLIC */
-		tp = SearchSysCache2(USERMAPPINGUSERSERVER,
-							 ObjectIdGetDatum(InvalidOid),
-							 ObjectIdGetDatum(serverid));
-	}
-
-	if (!HeapTupleIsValid(tp))
-		ereport(ERROR,
-				(errcode(ERRCODE_UNDEFINED_OBJECT),
-				 errmsg("user mapping not found for \"%s\"",
-						MappingUserName(userid))));
+	tp = find_user_mapping(userid, serverid);
 
 	um = (UserMapping *) palloc(sizeof(UserMapping));
 	um->umid = HeapTupleGetOid(tp);
@@ -212,6 +197,61 @@ GetUserMapping(Oid userid, Oid serverid)
 	return um;
 }
 
+/*
+ * GetUserMappingId - look up the user mapping, and return its OID
+ *
+ * If no mapping is found for the supplied user, we also look for
+ * PUBLIC mappings (userid == InvalidOid).
+ */
+Oid
+GetUserMappingId(Oid userid, Oid serverid)
+{
+	HeapTuple	tp;
+	Oid			umid;
+
+	tp = find_user_mapping(userid, serverid);
+
+	/* Extract the Oid */
+	umid = HeapTupleGetOid(tp);
+
+	ReleaseSysCache(tp);
+
+	return umid;
+}
+
+
+/*
+ * find_user_mapping - Guts of GetUserMapping family.
+ *
+ * If no mapping is found for the supplied user, we also look for
+ * PUBLIC mappings (userid == InvalidOid).
+ */
+static HeapTuple
+find_user_mapping(Oid userid, Oid serverid)
+{
+	HeapTuple	tp;
+
+	tp = SearchSysCache2(USERMAPPINGUSERSERVER,
+						 ObjectIdGetDatum(userid),
+						 ObjectIdGetDatum(serverid));
+
+	if (HeapTupleIsValid(tp))
+		return tp;
+
+	/* Not found for the specific user -- try PUBLIC */
+	tp = SearchSysCache2(USERMAPPINGUSERSERVER,
+							 ObjectIdGetDatum(InvalidOid),
+							 ObjectIdGetDatum(serverid));
+
+	if (!HeapTupleIsValid(tp))
+		ereport(ERROR,
+				(errcode(ERRCODE_UNDEFINED_OBJECT),
+				 errmsg("user mapping not found for \"%s\"",
+						MappingUserName(userid))));
+
+	return tp;
+}
+
 
 /*
  * GetForeignTable - look up the foreign table definition by relation oid.
diff --git a/src/backend/nodes/copyfuncs.c b/src/backend/nodes/copyfuncs.c
index 5877037df4c..a8b79fa8c31 100644
--- a/src/backend/nodes/copyfuncs.c
+++ b/src/backend/nodes/copyfuncs.c
@@ -95,6 +95,7 @@ _copyPlannedStmt(const PlannedStmt *from)
 	COPY_SCALAR_FIELD(nParamExec);
 	COPY_SCALAR_FIELD(hasRowSecurity);
 	COPY_SCALAR_FIELD(parallelModeNeeded);
+	COPY_SCALAR_FIELD(hasForeignJoin);
 
 	return newnode;
 }
diff --git a/src/backend/nodes/outfuncs.c b/src/backend/nodes/outfuncs.c
index b5e0b5578f1..b487c002a8c 100644
--- a/src/backend/nodes/outfuncs.c
+++ b/src/backend/nodes/outfuncs.c
@@ -259,6 +259,7 @@ _outPlannedStmt(StringInfo str, const PlannedStmt *node)
 	WRITE_INT_FIELD(nParamExec);
 	WRITE_BOOL_FIELD(hasRowSecurity);
 	WRITE_BOOL_FIELD(parallelModeNeeded);
+	WRITE_BOOL_FIELD(hasForeignJoin);
 }
 
 /*
@@ -1825,6 +1826,7 @@ _outPlannerGlobal(StringInfo str, const PlannerGlobal *node)
 	WRITE_BOOL_FIELD(hasRowSecurity);
 	WRITE_BOOL_FIELD(parallelModeOK);
 	WRITE_BOOL_FIELD(parallelModeNeeded);
+	WRITE_BOOL_FIELD(hasForeignJoin);
 }
 
 static void
diff --git a/src/backend/nodes/readfuncs.c b/src/backend/nodes/readfuncs.c
index a67b3370da0..6c461513d64 100644
--- a/src/backend/nodes/readfuncs.c
+++ b/src/backend/nodes/readfuncs.c
@@ -1396,6 +1396,7 @@ _readPlannedStmt(void)
 	READ_INT_FIELD(nParamExec);
 	READ_BOOL_FIELD(hasRowSecurity);
 	READ_BOOL_FIELD(parallelModeNeeded);
+	READ_BOOL_FIELD(hasForeignJoin);
 
 	READ_DONE();
 }
diff --git a/src/backend/optimizer/plan/createplan.c b/src/backend/optimizer/plan/createplan.c
index fda4df64210..bdac0b1860b 100644
--- a/src/backend/optimizer/plan/createplan.c
+++ b/src/backend/optimizer/plan/createplan.c
@@ -2152,6 +2152,15 @@ create_foreignscan_plan(PlannerInfo *root, ForeignPath *best_path,
 	scan_plan->fs_relids = best_path->path.parent->relids;
 
 	/*
+	 * If a join between foreign relations was pushed down, remember it. The
+	 * push-down safety of the join depends upon the server and user mapping
+	 * being same. That can change between planning and execution time, in which
+	 * case the plan should be invalidated.
+	 */
+	if (scan_relid == 0)
+		root->glob->hasForeignJoin = true;
+
+	/*
 	 * Replace any outer-relation variables with nestloop params in the qual,
 	 * fdw_exprs and fdw_recheck_quals expressions.  We do this last so that
 	 * the FDW doesn't have to be involved.  (Note that parts of fdw_exprs
diff --git a/src/backend/optimizer/plan/planner.c b/src/backend/optimizer/plan/planner.c
index c0ec905eb3f..a09b4b5b479 100644
--- a/src/backend/optimizer/plan/planner.c
+++ b/src/backend/optimizer/plan/planner.c
@@ -200,6 +200,7 @@ standard_planner(Query *parse, int cursorOptions, ParamListInfo boundParams)
 	glob->lastPlanNodeId = 0;
 	glob->transientPlan = false;
 	glob->hasRowSecurity = false;
+	glob->hasForeignJoin = false;
 
 	/*
 	 * Assess whether it's feasible to use parallel mode for this query. We
@@ -346,6 +347,7 @@ standard_planner(Query *parse, int cursorOptions, ParamListInfo boundParams)
 	result->nParamExec = glob->nParamExec;
 	result->hasRowSecurity = glob->hasRowSecurity;
 	result->parallelModeNeeded = glob->parallelModeNeeded;
+	result->hasForeignJoin = glob->hasForeignJoin;
 
 	return result;
 }
diff --git a/src/backend/optimizer/util/relnode.c b/src/backend/optimizer/util/relnode.c
index 7428c18af9f..420692f7a4d 100644
--- a/src/backend/optimizer/util/relnode.c
+++ b/src/backend/optimizer/util/relnode.c
@@ -14,6 +14,9 @@
  */
 #include "postgres.h"
 
+#include "miscadmin.h"
+#include "catalog/pg_class.h"
+#include "foreign/foreign.h"
 #include "optimizer/clauses.h"
 #include "optimizer/cost.h"
 #include "optimizer/pathnode.h"
@@ -127,6 +130,7 @@ build_simple_rel(PlannerInfo *root, int relid, RelOptKind reloptkind)
 	rel->subroot = NULL;
 	rel->subplan_params = NIL;
 	rel->serverid = InvalidOid;
+	rel->umid = InvalidOid;
 	rel->fdwroutine = NULL;
 	rel->fdw_private = NULL;
 	rel->baserestrictinfo = NIL;
@@ -166,6 +170,26 @@ build_simple_rel(PlannerInfo *root, int relid, RelOptKind reloptkind)
 			break;
 	}
 
+	/* For foreign tables get the user mapping */
+	if (rte->relkind == RELKIND_FOREIGN_TABLE)
+	{
+		/*
+		 * This should match what ExecCheckRTEPerms() does.
+		 *
+		 * Note that if the plan ends up depending on the user OID in any
+		 * way - e.g. if it depends on the computed user mapping OID - we must
+		 * ensure that it gets invalidated in the case of a user OID change.
+		 * See RevalidateCachedQuery and more generally the hasForeignJoin
+		 * flags in PlannerGlobal and PlannedStmt.
+		 */
+		Oid		userid;
+
+		userid = OidIsValid(rte->checkAsUser) ? rte->checkAsUser : GetUserId();
+		rel->umid = GetUserMappingId(userid, rel->serverid);
+	}
+	else
+		rel->umid = InvalidOid;
+
 	/* Save the finished struct in the query's simple_rel_array */
 	root->simple_rel_array[relid] = rel;
 
@@ -398,6 +422,7 @@ build_join_rel(PlannerInfo *root,
 	joinrel->subroot = NULL;
 	joinrel->subplan_params = NIL;
 	joinrel->serverid = InvalidOid;
+	joinrel->umid = InvalidOid;
 	joinrel->fdwroutine = NULL;
 	joinrel->fdw_private = NULL;
 	joinrel->baserestrictinfo = NIL;
@@ -408,12 +433,19 @@ build_join_rel(PlannerInfo *root,
 
 	/*
 	 * Set up foreign-join fields if outer and inner relation are foreign
-	 * tables (or joins) belonging to the same server.
+	 * tables (or joins) belonging to the same server and using the same
+	 * user mapping.
+	 *
+	 * Otherwise those fields are left invalid, so FDW API will not be called
+	 * for the join relation.
 	 */
 	if (OidIsValid(outer_rel->serverid) &&
-		inner_rel->serverid == outer_rel->serverid)
+		inner_rel->serverid == outer_rel->serverid &&
+		inner_rel->umid == outer_rel->umid)
 	{
+		Assert(OidIsValid(outer_rel->umid));
 		joinrel->serverid = outer_rel->serverid;
+		joinrel->umid = outer_rel->umid;
 		joinrel->fdwroutine = outer_rel->fdwroutine;
 	}
 
diff --git a/src/backend/utils/cache/plancache.c b/src/backend/utils/cache/plancache.c
index 539f4b9240c..a93825d0087 100644
--- a/src/backend/utils/cache/plancache.c
+++ b/src/backend/utils/cache/plancache.c
@@ -104,6 +104,8 @@ static TupleDesc PlanCacheComputeResultDesc(List *stmt_list);
 static void PlanCacheRelCallback(Datum arg, Oid relid);
 static void PlanCacheFuncCallback(Datum arg, int cacheid, uint32 hashvalue);
 static void PlanCacheSysCallback(Datum arg, int cacheid, uint32 hashvalue);
+static void PlanCacheUserMappingCallback(Datum arg, int cacheid,
+										 uint32 hashvalue);
 
 
 /*
@@ -119,6 +121,8 @@ InitPlanCache(void)
 	CacheRegisterSyscacheCallback(NAMESPACEOID, PlanCacheSysCallback, (Datum) 0);
 	CacheRegisterSyscacheCallback(OPEROID, PlanCacheSysCallback, (Datum) 0);
 	CacheRegisterSyscacheCallback(AMOPOPID, PlanCacheSysCallback, (Datum) 0);
+	/* User mapping change may invalidate plans with pushed down foreign join */
+	CacheRegisterSyscacheCallback(USERMAPPINGOID, PlanCacheUserMappingCallback, (Datum) 0);
 }
 
 /*
@@ -574,7 +578,8 @@ RevalidateCachedQuery(CachedPlanSource *plansource)
 	/*
 	 * If this is a new cached plan, then set the user id it was planned by
 	 * and under what row security settings; these are needed to determine
-	 * plan invalidation when RLS is involved.
+	 * plan invalidation when RLS is involved or foreign joins are pushed
+	 * down.
 	 */
 	if (!OidIsValid(plansource->planUserId))
 	{
@@ -610,6 +615,18 @@ RevalidateCachedQuery(CachedPlanSource *plansource)
 		plansource->is_valid = false;
 
 	/*
+	 * If we have a join pushed down to the foreign server and the current user
+	 * is different from the one for which the plan was created, invalidate the
+	 * generic plan since user mapping for the new user might make the join
+	 * unsafe to push down, or change which user mapping is used.
+	 */
+	if (plansource->is_valid &&
+		plansource->gplan &&
+		plansource->gplan->has_foreign_join &&
+		plansource->planUserId != GetUserId())
+		plansource->gplan->is_valid = false;
+
+	/*
 	 * If the query is currently valid, acquire locks on the referenced
 	 * objects; then check again.  We need to do it this way to cover the race
 	 * condition that an invalidation message arrives before we get the locks.
@@ -881,6 +898,7 @@ BuildCachedPlan(CachedPlanSource *plansource, List *qlist,
 	bool		spi_pushed;
 	MemoryContext plan_context;
 	MemoryContext oldcxt = CurrentMemoryContext;
+	ListCell	*lc;
 
 	/*
 	 * Normally the querytree should be valid already, but if it's not,
@@ -988,6 +1006,20 @@ BuildCachedPlan(CachedPlanSource *plansource, List *qlist,
 	plan->is_saved = false;
 	plan->is_valid = true;
 
+	/*
+	 * Walk through the plist and set hasForeignJoin if any of the plans have
+	 * it set.
+	 */
+	plan->has_foreign_join = false;
+	foreach(lc, plist)
+	{
+		PlannedStmt	*plan_stmt = (PlannedStmt *) lfirst(lc);
+
+		if (IsA(plan_stmt, PlannedStmt))
+			plan->has_foreign_join =
+				plan->has_foreign_join || plan_stmt->hasForeignJoin;
+	}
+
 	/* assign generation number to new plan */
 	plan->generation = ++(plansource->generation);
 
@@ -1844,6 +1876,40 @@ PlanCacheSysCallback(Datum arg, int cacheid, uint32 hashvalue)
 }
 
 /*
+ * PlanCacheUserMappingCallback
+ * 		Syscache inval callback function for user mapping cache invalidation.
+ *
+ * 	Invalidates plans which have pushed down foreign joins.
+ */
+static void
+PlanCacheUserMappingCallback(Datum arg, int cacheid, uint32 hashvalue)
+{
+	CachedPlanSource *plansource;
+
+	for (plansource = first_saved_plan; plansource; plansource = plansource->next_saved)
+	{
+		Assert(plansource->magic == CACHEDPLANSOURCE_MAGIC);
+
+		/* No work if it's already invalidated */
+		if (!plansource->is_valid)
+			continue;
+
+		/* Never invalidate transaction control commands */
+		if (IsTransactionStmtPlan(plansource))
+			continue;
+
+		/*
+		 * If the plan has pushed down foreign joins, those join may become
+		 * unsafe to push down because of user mapping changes. Invalidate only
+		 * the generic plan, since changes to user mapping do not invalidate the
+		 * parse tree.
+		 */
+		if (plansource->gplan && plansource->gplan->has_foreign_join)
+			plansource->gplan->is_valid = false;
+	}
+}
+
+/*
  * ResetPlanCache: invalidate all cached plans.
  */
 void
diff --git a/src/include/foreign/foreign.h b/src/include/foreign/foreign.h
index 5dc2c90f3c3..d1359163e48 100644
--- a/src/include/foreign/foreign.h
+++ b/src/include/foreign/foreign.h
@@ -72,6 +72,7 @@ typedef struct ForeignTable
 extern ForeignServer *GetForeignServer(Oid serverid);
 extern ForeignServer *GetForeignServerByName(const char *name, bool missing_ok);
 extern UserMapping *GetUserMapping(Oid userid, Oid serverid);
+extern Oid GetUserMappingId(Oid userid, Oid serverid);
 extern ForeignDataWrapper *GetForeignDataWrapper(Oid fdwid);
 extern ForeignDataWrapper *GetForeignDataWrapperByName(const char *name,
 							bool missing_ok);
diff --git a/src/include/nodes/plannodes.h b/src/include/nodes/plannodes.h
index e823c830115..55d6bbe8f0e 100644
--- a/src/include/nodes/plannodes.h
+++ b/src/include/nodes/plannodes.h
@@ -73,6 +73,7 @@ typedef struct PlannedStmt
 	bool		hasRowSecurity; /* row security applied? */
 
 	bool		parallelModeNeeded; /* parallel mode required to execute? */
+	bool		hasForeignJoin;	/* Plan has a pushed down foreign join */
 } PlannedStmt;
 
 /* macro for fetching the Plan associated with a SubPlan node */
diff --git a/src/include/nodes/relation.h b/src/include/nodes/relation.h
index b233b62d56c..94925984bf2 100644
--- a/src/include/nodes/relation.h
+++ b/src/include/nodes/relation.h
@@ -108,6 +108,7 @@ typedef struct PlannerGlobal
 	bool		parallelModeOK; /* parallel mode potentially OK? */
 
 	bool		parallelModeNeeded;		/* parallel mode actually required? */
+	bool		hasForeignJoin;	/* does have a pushed down foreign join */
 } PlannerGlobal;
 
 /* macro for fetching the Plan associated with a SubPlan node */
@@ -490,6 +491,7 @@ typedef struct RelOptInfo
 
 	/* Information about foreign tables and foreign joins */
 	Oid			serverid;		/* identifies server for the table or join */
+	Oid			umid;			/* identifies user mapping for the table or join */
 	/* use "struct FdwRoutine" to avoid including fdwapi.h here */
 	struct FdwRoutine *fdwroutine;
 	void	   *fdw_private;
diff --git a/src/include/utils/plancache.h b/src/include/utils/plancache.h
index 0929f58d6b0..7a98c5fa975 100644
--- a/src/include/utils/plancache.h
+++ b/src/include/utils/plancache.h
@@ -135,6 +135,7 @@ typedef struct CachedPlan
 								 * changes from this value */
 	int			generation;		/* parent's generation number for this plan */
 	int			refcount;		/* count of live references to this struct */
+	bool		has_foreign_join; /* plan has pushed down a foreign join */
 	MemoryContext context;		/* context containing this CachedPlan */
 } CachedPlan;