Test

src/backend/distributed/executor/tuple_destination.c

@@ -181,7 +181,7 @@ EnsureIntermediateSizeLimitNotExceeded(TupleDestinationStats *tupleDestinationSt
 static TupleDesc
 TupleStoreTupleDestTupleDescForQuery(TupleDestination *self, int queryNumber)
 {
-	Assert(queryNumber == 0);
+	/*Assert(queryNumber == 0); */
 
 	TupleStoreTupleDestination *tupleDest = (TupleStoreTupleDestination *) self;
 

src/backend/distributed/metadata/metadata_sync.c

@@ -122,7 +122,6 @@ static List * GrantOnSequenceDDLCommands(Oid sequenceOid);
 static List * GenerateGrantOnSequenceQueriesFromAclItem(Oid sequenceOid,
 														AclItem *aclItem);
 static void SetLocalReplicateReferenceTablesOnActivate(bool state);
-static char * GenerateSetRoleQuery(Oid roleOid);
 static void MetadataSyncSigTermHandler(SIGNAL_ARGS);
 static void MetadataSyncSigAlrmHandler(SIGNAL_ARGS);
 
@@ -2484,7 +2483,7 @@ SetLocalReplicateReferenceTablesOnActivate(bool state)
 }
 
 
-static char *
+char *
 GenerateSetRoleQuery(Oid roleOid)
 {
 	StringInfo buf = makeStringInfo();

src/backend/distributed/planner/distributed_planner.c

@@ -1156,6 +1156,28 @@ CreateDistributedPlan(uint64 planId, Query *originalQuery, Query *query, ParamLi
 	 */
 	CheckNodeIsDumpable((Node *) logicalPlan);
 
+	RTEListProperties *checkForViews = GetRTEListPropertiesForQuery(originalQuery);
+
+	if (checkForViews->hasView)
+	{
+		HeapTuple tuple = SearchSysCache1(RELOID, ObjectIdGetDatum(
+											  checkForViews->viewId));
+		if (!HeapTupleIsValid(tuple))
+		{
+			ereport(ERROR,
+					(errcode(ERRCODE_UNDEFINED_TABLE),
+					 errmsg("relation with OID %u does not exist",
+							checkForViews->viewId)));
+		}
+
+		Oid ownerId = ((Form_pg_class) GETSTRUCT(tuple))->relowner;
+
+		ReleaseSysCache(tuple);
+
+		plannerRestrictionContext->relationRestrictionContext->hasView = true;
+		plannerRestrictionContext->relationRestrictionContext->viewOwnerId = ownerId;
+	}
+
 	/* Create the physical plan */
 	distributedPlan = CreatePhysicalDistributedPlan(logicalPlan,
 													plannerRestrictionContext);
@@ -2497,6 +2519,8 @@ GetRTEListProperties(List *rangeTableList)
 			 * Skip over views, distributed tables within (regular) views are
 			 * already in rangeTableList.
 			 */
+			rteListProperties->hasView = true;
+			rteListProperties->viewId = rangeTableEntry->relid;
 			continue;
 		}
 

src/backend/distributed/planner/multi_physical_planner.c

@@ -42,6 +42,7 @@
 #include "distributed/coordinator_protocol.h"
 #include "distributed/intermediate_results.h"
 #include "distributed/metadata_cache.h"
+#include "distributed/metadata_sync.h"
 #include "distributed/multi_router_planner.h"
 #include "distributed/multi_join_order.h"
 #include "distributed/multi_logical_optimizer.h"
@@ -2074,6 +2075,11 @@ BuildJobTreeTaskList(Job *jobTree, PlannerRestrictionContext *plannerRestriction
 	for (int32 jobIndex = (flattenedJobCount - 1); jobIndex >= 0; jobIndex--)
 	{
 		Job *job = (Job *) list_nth(flattenedJobList, jobIndex);
+		if (plannerRestrictionContext->relationRestrictionContext->hasView)
+		{
+			job->userId =
+				plannerRestrictionContext->relationRestrictionContext->viewOwnerId;
+		}
 		List *sqlTaskList = NIL;
 		ListCell *assignedSqlTaskCell = NULL;
 
@@ -2713,6 +2719,12 @@ SqlTaskList(Job *job)
 
 		Task *sqlTask = CreateBasicTask(jobId, taskIdIndex, READ_TASK,
 										sqlQueryString->data);
+		if (job->userId != InvalidOid)
+		{
+			List *cmds = list_make2(GenerateSetRoleQuery(job->userId),
+									sqlQueryString->data);
+			SetTaskQueryStringList(sqlTask, cmds);
+		}
 		sqlTask->dependentTaskList = dataFetchTaskList;
 		sqlTask->relationShardList = BuildRelationShardList(fragmentRangeTableList,
 															fragmentCombination);

src/include/distributed/distributed_planner.h

@@ -36,6 +36,8 @@ extern int PlannerLevel;
 
 typedef struct RelationRestrictionContext
 {
+	bool hasView;
+	Oid viewOwnerId;
 	bool allReferenceTables;
 	List *relationRestrictionList;
 } RelationRestrictionContext;
@@ -148,6 +150,9 @@ typedef struct RTEListProperties
 	bool hasCitusTable;
 
 	bool hasMaterializedView;
+
+	bool hasView;
+	Oid viewId;
 } RTEListProperties;
 
 

src/include/distributed/metadata_sync.h

@@ -110,6 +110,7 @@ extern void SyncNewColocationGroupToNodes(uint32 colocationId, int shardCount,
 										  Oid distributionColumType,
 										  Oid distributionColumnCollation);
 extern void SyncDeleteColocationGroupToNodes(uint32 colocationId);
+extern char * GenerateSetRoleQuery(Oid roleOid);
 
 #define DELETE_ALL_NODES "DELETE FROM pg_dist_node"
 #define DELETE_ALL_PLACEMENTS "DELETE FROM pg_dist_placement"

src/include/distributed/multi_physical_planner.h

@@ -136,6 +136,7 @@ typedef struct Job
 	CitusNode type;
 	uint64 jobId;
 	Query *jobQuery;
+	Oid userId;
 	List *taskList;
 	List *dependentJobList;
 	bool subqueryPushdown;