Only try to push down foreign joins if the user mapping OIDs match.

Previously, the foreign join pushdown infrastructure left the question
of security entirely up to individual FDWs, but it would be easy for
a foreign data wrapper to inadvertently open up subtle security holes
that way.  So, make it the core code's job to determine which user
mapping OID is relevant, and don't attempt join pushdown unless it's
the same for all relevant relations.

Per a suggestion from Tom Lane.  Shigeru Hanada and Ashutosh Bapat,
reviewed by Etsuro Fujita and KaiGai Kohei, with some further
changes by me.

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

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.

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

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

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

src/backend/optimizer/plan/createplan.c

@@ -2151,6 +2151,15 @@ create_foreignscan_plan(PlannerInfo *root, ForeignPath *best_path,
 	/* Likewise, copy the relids that are represented by this foreign scan */
 	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

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

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

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))
 	{
@@ -609,6 +614,18 @@ RevalidateCachedQuery(CachedPlanSource *plansource)
 			|| plansource->row_security_env != row_security))
 		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
@@ -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);
 
@@ -1843,6 +1875,40 @@ PlanCacheSysCallback(Datum arg, int cacheid, uint32 hashvalue)
 	ResetPlanCache();
 }
 
+/*
+ * 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.
  */

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

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

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;

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;