Improve error semantics for INSERT..SELECT

With this commit, we error out if a worker query cannot be executed
on all placements of a target insert shard interval.

src/backend/distributed/executor/multi_router_executor.c

@@ -1027,6 +1027,7 @@ ExecuteModifyTasks(List *taskList, bool expectResults, ParamListInfo paramListIn
 			if (placementIndex >= list_length(connectionList))
 			{
 				/* no more active placements for this task */
+				taskIndex++;
 				continue;
 			}
 

src/backend/distributed/planner/multi_router_planner.c

@@ -104,7 +104,7 @@ static Task * RouterSelectTask(Query *originalQuery,
 static bool RouterSelectQuery(Query *originalQuery,
 							  RelationRestrictionContext *restrictionContext,
 							  List **placementList, uint64 *anchorShardId,
-							  List **selectShardList);
+							  List **selectShardList, bool replacePrunedQueryWithDummy);
 static List * TargetShardIntervalsForSelect(Query *query,
 											RelationRestrictionContext *restrictionContext);
 static List * WorkersContainingAllShards(List *prunedShardIntervalsList);
@@ -343,6 +343,7 @@ RouterModifyTaskForShardInterval(Query *originalQuery, ShardInterval *shardInter
 	List *intersectedPlacementList = NULL;
 	bool routerPlannable = false;
 	bool upsertQuery = false;
+	bool replacePrunedQueryWithDummy = false;
 
 	/* grab shared metadata lock to stop concurrent placement additions */
 	LockShardDistributionMetadata(shardId, ShareLock);
@@ -372,6 +373,9 @@ RouterModifyTaskForShardInterval(Query *originalQuery, ShardInterval *shardInter
 	 */
 	AddShardIntervalRestrictionToSelect(copiedSubquery, shardInterval);
 
+	/* mark that we don't want the router planner to generate dummy hosts/queries */
+	replacePrunedQueryWithDummy = false;
+
 	/*
 	 * Use router select planner to decide on whether we can push down the query
 	 * or not. If we can, we also rely on the side-effects that all RTEs have been
@@ -379,7 +383,7 @@ RouterModifyTaskForShardInterval(Query *originalQuery, ShardInterval *shardInter
 	 */
 	routerPlannable = RouterSelectQuery(copiedSubquery, copiedRestrictionContext,
 										&selectPlacementList, &selectAnchorShardId,
-										&selectShardList);
+										&selectShardList, replacePrunedQueryWithDummy);
 
 	if (!routerPlannable)
 	{
@@ -389,24 +393,35 @@ RouterModifyTaskForShardInterval(Query *originalQuery, ShardInterval *shardInter
 						errdetail("Select query cannot be pushed down to the worker.")));
 	}
 
-	/* Ensure that we have INSERTed table's placement exists on the same worker */
+
+	/* ensure that we do not send queries where select is pruned away completely */
+	if (list_length(selectPlacementList) == 0)
+	{
+		ereport(DEBUG2, (errmsg("Skipping target shard interval %ld since "
+								"SELECT query for it pruned away", shardId)));
+
+		return NULL;
+	}
+
+	/* get the placements for insert target shard and its intersection with select */
 	insertShardPlacementList = FinalizedShardPlacementList(shardId);
 	intersectedPlacementList = IntersectPlacementList(insertShardPlacementList,
 													  selectPlacementList);
 
+	/*
+	 * If insert target does not have exactly the same placements with the select,
+	 * we sholdn't run the query.
+	 */
 	if (list_length(insertShardPlacementList) != list_length(intersectedPlacementList))
 	{
-		ereport(DEBUG2, (errmsg("could not generate task for target shardId: %ld",
+		ereport(ERROR, (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
-								shardId),
+						errmsg("cannot perform distributed planning for the given "
-						 errdetail("Insert query hits %d placements, Select query "
+							   "modification"),
-								   "hits %d placements and only %d of those placements match.",
+						errdetail("Insert query cannot be executed on all placements "
-								   list_length(insertShardPlacementList),
+								  "for shard %ld", shardId)));
-								   list_length(selectPlacementList),
-								   list_length(intersectedPlacementList))));
-
-		return NULL;
 	}
 
+
 	/* this is required for correct deparsing of the query */
 	ReorderInsertSelectTargetLists(copiedQuery, copiedInsertRte, copiedSubqueryRte);
 
@@ -1817,9 +1832,14 @@ RouterSelectTask(Query *originalQuery, RelationRestrictionContext *restrictionCo
 	bool upsertQuery = false;
 	uint64 shardId = INVALID_SHARD_ID;
 	List *selectShardList = NIL;
+	bool replacePrunedQueryWithDummy = false;
+
+	/* router planner should create task even if it deosn't hit a shard at all */
+	replacePrunedQueryWithDummy = true;
 
 	queryRoutable = RouterSelectQuery(originalQuery, restrictionContext,
-									  placementList, &shardId, &selectShardList);
+									  placementList, &shardId, &selectShardList,
+									  replacePrunedQueryWithDummy);
 
 
 	if (!queryRoutable)
@@ -1853,7 +1873,8 @@ RouterSelectTask(Query *originalQuery, RelationRestrictionContext *restrictionCo
  */
 static bool
 RouterSelectQuery(Query *originalQuery, RelationRestrictionContext *restrictionContext,
-				  List **placementList, uint64 *anchorShardId, List **selectShardList)
+				  List **placementList, uint64 *anchorShardId, List **selectShardList,
+				  bool replacePrunedQueryWithDummy)
 {
 	List *prunedRelationShardList = TargetShardIntervalsForSelect(originalQuery,
 																  restrictionContext);
@@ -1901,14 +1922,15 @@ RouterSelectQuery(Query *originalQuery, RelationRestrictionContext *restrictionC
 
 	/*
 	 * Determine the worker that has all shard placements if a shard placement found.
-	 * If no shard placement exists, we will still run the query but the result will
+	 * If no shard placement exists and replacePrunedQueryWithDummy flag is set, we will
-	 * be empty. We create a dummy shard placement for the first active worker.
+	 * still run the query but the result will be empty. We create a dummy shard
+	 * placement for the first active worker.
 	 */
 	if (shardsPresent)
 	{
 		workerList = WorkersContainingAllShards(prunedRelationShardList);
 	}
-	else
+	else if (replacePrunedQueryWithDummy)
 	{
 		List *workerNodeList = WorkerNodeList();
 		if (workerNodeList != NIL)
@@ -1922,6 +1944,17 @@ RouterSelectQuery(Query *originalQuery, RelationRestrictionContext *restrictionC
 			workerList = lappend(workerList, dummyPlacement);
 		}
 	}
+	else
+	{
+		/*
+		 * For INSERT ... SELECT, this query could be still a valid for some other target
+		 * shard intervals. Thus, we should return empty list if there aren't any matching
+		 * workers, so that the caller can decide what to do with this task.
+		 */
+		workerList = NIL;
+
+		return true;
+	}
 
 	if (workerList == NIL)
 	{

src/test/regress/expected/multi_insert_select.out

@@ -131,8 +131,7 @@ DEBUG:  predicate pruning for shardId 13300000
 DEBUG:  predicate pruning for shardId 13300001
 DEBUG:  predicate pruning for shardId 13300002
 DEBUG:  predicate pruning for shardId 13300003
-DEBUG:  could not generate task for target shardId: 13300004
+DEBUG:  Skipping target shard interval 13300004 since SELECT query for it pruned away
-DETAIL:  Insert query hits 2 placements, Select query hits 1 placements and only 1 of those placements match.
 DEBUG:  predicate pruning for shardId 13300000
 DEBUG:  predicate pruning for shardId 13300002
 DEBUG:  predicate pruning for shardId 13300003
@@ -141,14 +140,12 @@ DEBUG:  predicate pruning for shardId 13300000
 DEBUG:  predicate pruning for shardId 13300001
 DEBUG:  predicate pruning for shardId 13300002
 DEBUG:  predicate pruning for shardId 13300003
-DEBUG:  could not generate task for target shardId: 13300006
+DEBUG:  Skipping target shard interval 13300006 since SELECT query for it pruned away
-DETAIL:  Insert query hits 2 placements, Select query hits 1 placements and only 1 of those placements match.
 DEBUG:  predicate pruning for shardId 13300000
 DEBUG:  predicate pruning for shardId 13300001
 DEBUG:  predicate pruning for shardId 13300002
 DEBUG:  predicate pruning for shardId 13300003
-DEBUG:  could not generate task for target shardId: 13300007
+DEBUG:  Skipping target shard interval 13300007 since SELECT query for it pruned away
-DETAIL:  Insert query hits 2 placements, Select query hits 1 placements and only 1 of those placements match.
 DEBUG:  ProcessQuery
 DEBUG:  Plan is router executable
 DEBUG:  CommitTransactionCommand
@@ -185,20 +182,17 @@ DEBUG:  predicate pruning for shardId 13300000
 DEBUG:  predicate pruning for shardId 13300001
 DEBUG:  predicate pruning for shardId 13300002
 DEBUG:  predicate pruning for shardId 13300003
-DEBUG:  could not generate task for target shardId: 13300005
+DEBUG:  Skipping target shard interval 13300005 since SELECT query for it pruned away
-DETAIL:  Insert query hits 2 placements, Select query hits 1 placements and only 1 of those placements match.
 DEBUG:  predicate pruning for shardId 13300000
 DEBUG:  predicate pruning for shardId 13300001
 DEBUG:  predicate pruning for shardId 13300002
 DEBUG:  predicate pruning for shardId 13300003
-DEBUG:  could not generate task for target shardId: 13300006
+DEBUG:  Skipping target shard interval 13300006 since SELECT query for it pruned away
-DETAIL:  Insert query hits 2 placements, Select query hits 1 placements and only 1 of those placements match.
 DEBUG:  predicate pruning for shardId 13300000
 DEBUG:  predicate pruning for shardId 13300001
 DEBUG:  predicate pruning for shardId 13300002
 DEBUG:  predicate pruning for shardId 13300003
-DEBUG:  could not generate task for target shardId: 13300007
+DEBUG:  Skipping target shard interval 13300007 since SELECT query for it pruned away
-DETAIL:  Insert query hits 2 placements, Select query hits 1 placements and only 1 of those placements match.
 DEBUG:  ProcessQuery
 DEBUG:  Plan is router executable
 DEBUG:  CommitTransactionCommand
@@ -215,14 +209,10 @@ WHERE
 DEBUG:  StartTransactionCommand
 DEBUG:  StartTransaction
 DEBUG:  name: unnamed; blockState:       DEFAULT; state: INPROGR, xid/subid/cid: 0/1/0, nestlvl: 1, children: 
-DEBUG:  could not generate task for target shardId: 13300004
+DEBUG:  Skipping target shard interval 13300004 since SELECT query for it pruned away
-DETAIL:  Insert query hits 2 placements, Select query hits 1 placements and only 1 of those placements match.
+DEBUG:  Skipping target shard interval 13300005 since SELECT query for it pruned away
-DEBUG:  could not generate task for target shardId: 13300005
+DEBUG:  Skipping target shard interval 13300006 since SELECT query for it pruned away
-DETAIL:  Insert query hits 2 placements, Select query hits 1 placements and only 1 of those placements match.
+DEBUG:  Skipping target shard interval 13300007 since SELECT query for it pruned away
-DEBUG:  could not generate task for target shardId: 13300006
-DETAIL:  Insert query hits 2 placements, Select query hits 1 placements and only 1 of those placements match.
-DEBUG:  could not generate task for target shardId: 13300007
-DETAIL:  Insert query hits 2 placements, Select query hits 1 placements and only 1 of those placements match.
 DEBUG:  ProcessQuery
 DEBUG:  Plan is router executable
 DEBUG:  CommitTransactionCommand
@@ -239,14 +229,10 @@ WHERE
 DEBUG:  StartTransactionCommand
 DEBUG:  StartTransaction
 DEBUG:  name: unnamed; blockState:       DEFAULT; state: INPROGR, xid/subid/cid: 0/1/0, nestlvl: 1, children: 
-DEBUG:  could not generate task for target shardId: 13300004
+DEBUG:  Skipping target shard interval 13300004 since SELECT query for it pruned away
-DETAIL:  Insert query hits 2 placements, Select query hits 1 placements and only 1 of those placements match.
+DEBUG:  Skipping target shard interval 13300005 since SELECT query for it pruned away
-DEBUG:  could not generate task for target shardId: 13300005
+DEBUG:  Skipping target shard interval 13300006 since SELECT query for it pruned away
-DETAIL:  Insert query hits 2 placements, Select query hits 1 placements and only 1 of those placements match.
+DEBUG:  Skipping target shard interval 13300007 since SELECT query for it pruned away
-DEBUG:  could not generate task for target shardId: 13300006
-DETAIL:  Insert query hits 2 placements, Select query hits 1 placements and only 1 of those placements match.
-DEBUG:  could not generate task for target shardId: 13300007
-DETAIL:  Insert query hits 2 placements, Select query hits 1 placements and only 1 of those placements match.
 DEBUG:  ProcessQuery
 DEBUG:  Plan is router executable
 DEBUG:  CommitTransactionCommand
@@ -326,8 +312,7 @@ DEBUG:  predicate pruning for shardId 13300000
 DEBUG:  predicate pruning for shardId 13300001
 DEBUG:  predicate pruning for shardId 13300002
 DEBUG:  predicate pruning for shardId 13300003
-DEBUG:  could not generate task for target shardId: 13300004
+DEBUG:  Skipping target shard interval 13300004 since SELECT query for it pruned away
-DETAIL:  Insert query hits 2 placements, Select query hits 1 placements and only 1 of those placements match.
 DEBUG:  predicate pruning for shardId 13300000
 DEBUG:  predicate pruning for shardId 13300002
 DEBUG:  predicate pruning for shardId 13300003
@@ -336,8 +321,7 @@ DEBUG:  predicate pruning for shardId 13300000
 DEBUG:  predicate pruning for shardId 13300001
 DEBUG:  predicate pruning for shardId 13300002
 DEBUG:  predicate pruning for shardId 13300003
-DEBUG:  could not generate task for target shardId: 13300006
+DEBUG:  Skipping target shard interval 13300006 since SELECT query for it pruned away
-DETAIL:  Insert query hits 2 placements, Select query hits 1 placements and only 1 of those placements match.
 DEBUG:  predicate pruning for shardId 13300000
 DEBUG:  predicate pruning for shardId 13300001
 DEBUG:  predicate pruning for shardId 13300002
@@ -1475,3 +1459,229 @@ DEBUG:  ProcessUtility
 CREATE VIEW test_view AS SELECT * FROM raw_events_first;
 INSERT INTO raw_events_second SELECT * FROM test_view;
 ERROR:  cannot plan queries that include both regular and partitioned relations
+-- we need this in our next test
+truncate raw_events_first;
+SET client_min_messages TO DEBUG4;
+DEBUG:  CommitTransactionCommand
+DEBUG:  CommitTransaction
+DEBUG:  name: unnamed; blockState:       STARTED; state: INPROGR, xid/subid/cid: 0/1/0, nestlvl: 1, children: 
+-- first show that the query works now
+INSERT INTO raw_events_first SELECT * FROM raw_events_second;
+DEBUG:  StartTransactionCommand
+DEBUG:  StartTransaction
+DEBUG:  name: unnamed; blockState:       DEFAULT; state: INPROGR, xid/subid/cid: 0/1/0, nestlvl: 1, children: 
+DEBUG:  predicate pruning for shardId 13300005
+DEBUG:  predicate pruning for shardId 13300006
+DEBUG:  predicate pruning for shardId 13300007
+DEBUG:  distributed statement: INSERT INTO public.raw_events_first_13300000 AS citus_table_alias (user_id, "time", value_1, value_2, value_3, value_4) SELECT user_id, "time", value_1, value_2, value_3, value_4 FROM public.raw_events_second_13300004 raw_events_second WHERE ((hashint4(user_id) >= '-2147483648'::integer) AND (hashint4(user_id) <= '-1073741825'::integer))
+DEBUG:  predicate pruning for shardId 13300004
+DEBUG:  predicate pruning for shardId 13300006
+DEBUG:  predicate pruning for shardId 13300007
+DEBUG:  distributed statement: INSERT INTO public.raw_events_first_13300001 AS citus_table_alias (user_id, "time", value_1, value_2, value_3, value_4) SELECT user_id, "time", value_1, value_2, value_3, value_4 FROM public.raw_events_second_13300005 raw_events_second WHERE ((hashint4(user_id) >= '-1073741824'::integer) AND (hashint4(user_id) <= '-1'::integer))
+DEBUG:  predicate pruning for shardId 13300004
+DEBUG:  predicate pruning for shardId 13300005
+DEBUG:  predicate pruning for shardId 13300007
+DEBUG:  distributed statement: INSERT INTO public.raw_events_first_13300002 AS citus_table_alias (user_id, "time", value_1, value_2, value_3, value_4) SELECT user_id, "time", value_1, value_2, value_3, value_4 FROM public.raw_events_second_13300006 raw_events_second WHERE ((hashint4(user_id) >= 0) AND (hashint4(user_id) <= 1073741823))
+DEBUG:  predicate pruning for shardId 13300004
+DEBUG:  predicate pruning for shardId 13300005
+DEBUG:  predicate pruning for shardId 13300006
+DEBUG:  distributed statement: INSERT INTO public.raw_events_first_13300003 AS citus_table_alias (user_id, "time", value_1, value_2, value_3, value_4) SELECT user_id, "time", value_1, value_2, value_3, value_4 FROM public.raw_events_second_13300007 raw_events_second WHERE ((hashint4(user_id) >= 1073741824) AND (hashint4(user_id) <= 2147483647))
+DEBUG:  ProcessQuery
+DEBUG:  Plan is router executable
+DEBUG:  CommitTransactionCommand
+DEBUG:  CommitTransaction
+DEBUG:  name: unnamed; blockState:       STARTED; state: INPROGR, xid/subid/cid: 0/1/0, nestlvl: 1, children: 
+DEBUG:  sent COMMIT over connection 13300001
+DEBUG:  sent COMMIT over connection 13300001
+DEBUG:  sent COMMIT over connection 13300000
+DEBUG:  sent COMMIT over connection 13300000
+DEBUG:  sent COMMIT over connection 13300002
+DEBUG:  sent COMMIT over connection 13300002
+DEBUG:  sent COMMIT over connection 13300003
+DEBUG:  sent COMMIT over connection 13300003
+SET client_min_messages TO INFO;
+DEBUG:  StartTransactionCommand
+DEBUG:  StartTransaction
+DEBUG:  name: unnamed; blockState:       DEFAULT; state: INPROGR, xid/subid/cid: 0/1/0, nestlvl: 1, children: 
+DEBUG:  ProcessUtility
+truncate raw_events_first;
+SET client_min_messages TO DEBUG4;
+DEBUG:  CommitTransactionCommand
+DEBUG:  CommitTransaction
+DEBUG:  name: unnamed; blockState:       STARTED; state: INPROGR, xid/subid/cid: 0/1/0, nestlvl: 1, children: 
+-- now show that it works for a single shard query as well
+INSERT INTO raw_events_first SELECT * FROM raw_events_second WHERE user_id = 5;
+DEBUG:  StartTransactionCommand
+DEBUG:  StartTransaction
+DEBUG:  name: unnamed; blockState:       DEFAULT; state: INPROGR, xid/subid/cid: 0/1/0, nestlvl: 1, children: 
+DEBUG:  predicate pruning for shardId 13300005
+DEBUG:  predicate pruning for shardId 13300006
+DEBUG:  predicate pruning for shardId 13300007
+DEBUG:  distributed statement: INSERT INTO public.raw_events_first_13300000 AS citus_table_alias (user_id, "time", value_1, value_2, value_3, value_4) SELECT user_id, "time", value_1, value_2, value_3, value_4 FROM public.raw_events_second_13300004 raw_events_second WHERE ((user_id = 5) AND ((hashint4(user_id) >= '-2147483648'::integer) AND (hashint4(user_id) <= '-1073741825'::integer)))
+DEBUG:  predicate pruning for shardId 13300004
+DEBUG:  predicate pruning for shardId 13300005
+DEBUG:  predicate pruning for shardId 13300006
+DEBUG:  predicate pruning for shardId 13300007
+DEBUG:  Skipping target shard interval 13300001 since SELECT query for it pruned away
+DEBUG:  predicate pruning for shardId 13300004
+DEBUG:  predicate pruning for shardId 13300005
+DEBUG:  predicate pruning for shardId 13300006
+DEBUG:  predicate pruning for shardId 13300007
+DEBUG:  Skipping target shard interval 13300002 since SELECT query for it pruned away
+DEBUG:  predicate pruning for shardId 13300004
+DEBUG:  predicate pruning for shardId 13300005
+DEBUG:  predicate pruning for shardId 13300006
+DEBUG:  predicate pruning for shardId 13300007
+DEBUG:  Skipping target shard interval 13300003 since SELECT query for it pruned away
+DEBUG:  ProcessQuery
+DEBUG:  Plan is router executable
+DEBUG:  CommitTransactionCommand
+DEBUG:  CommitTransaction
+DEBUG:  name: unnamed; blockState:       STARTED; state: INPROGR, xid/subid/cid: 0/1/0, nestlvl: 1, children: 
+SET client_min_messages TO INFO;
+DEBUG:  StartTransactionCommand
+DEBUG:  StartTransaction
+DEBUG:  name: unnamed; blockState:       DEFAULT; state: INPROGR, xid/subid/cid: 0/1/0, nestlvl: 1, children: 
+DEBUG:  ProcessUtility
+-- if a single shard of the SELECT is unhealty, the query should fail
+UPDATE pg_dist_shard_placement SET shardstate = 3 WHERE shardid = 13300004 AND nodeport = :worker_1_port;
+truncate raw_events_first;
+SET client_min_messages TO DEBUG4;
+DEBUG:  CommitTransactionCommand
+DEBUG:  CommitTransaction
+DEBUG:  name: unnamed; blockState:       STARTED; state: INPROGR, xid/subid/cid: 0/1/0, nestlvl: 1, children: 
+-- this should fail
+INSERT INTO raw_events_first SELECT * FROM raw_events_second;
+DEBUG:  StartTransactionCommand
+DEBUG:  StartTransaction
+DEBUG:  name: unnamed; blockState:       DEFAULT; state: INPROGR, xid/subid/cid: 0/1/0, nestlvl: 1, children: 
+DEBUG:  predicate pruning for shardId 13300005
+DEBUG:  predicate pruning for shardId 13300006
+DEBUG:  predicate pruning for shardId 13300007
+ERROR:  cannot perform distributed planning for the given modification
+DETAIL:  Insert query cannot be executed on all placements for shard 13300000
+-- this should also fail
+INSERT INTO raw_events_first SELECT * FROM raw_events_second WHERE user_id = 5;
+DEBUG:  StartTransactionCommand
+DEBUG:  StartTransaction
+DEBUG:  name: unnamed; blockState:       DEFAULT; state: INPROGR, xid/subid/cid: 0/1/0, nestlvl: 1, children: 
+DEBUG:  predicate pruning for shardId 13300005
+DEBUG:  predicate pruning for shardId 13300006
+DEBUG:  predicate pruning for shardId 13300007
+ERROR:  cannot perform distributed planning for the given modification
+DETAIL:  Insert query cannot be executed on all placements for shard 13300000
+-- but this should work given that it hits different shard
+INSERT INTO raw_events_first SELECT * FROM raw_events_second WHERE user_id = 6;
+DEBUG:  StartTransactionCommand
+DEBUG:  StartTransaction
+DEBUG:  name: unnamed; blockState:       DEFAULT; state: INPROGR, xid/subid/cid: 0/1/0, nestlvl: 1, children: 
+DEBUG:  predicate pruning for shardId 13300004
+DEBUG:  predicate pruning for shardId 13300005
+DEBUG:  predicate pruning for shardId 13300006
+DEBUG:  predicate pruning for shardId 13300007
+DEBUG:  Skipping target shard interval 13300000 since SELECT query for it pruned away
+DEBUG:  predicate pruning for shardId 13300004
+DEBUG:  predicate pruning for shardId 13300005
+DEBUG:  predicate pruning for shardId 13300006
+DEBUG:  predicate pruning for shardId 13300007
+DEBUG:  Skipping target shard interval 13300001 since SELECT query for it pruned away
+DEBUG:  predicate pruning for shardId 13300004
+DEBUG:  predicate pruning for shardId 13300005
+DEBUG:  predicate pruning for shardId 13300007
+DEBUG:  distributed statement: INSERT INTO public.raw_events_first_13300002 AS citus_table_alias (user_id, "time", value_1, value_2, value_3, value_4) SELECT user_id, "time", value_1, value_2, value_3, value_4 FROM public.raw_events_second_13300006 raw_events_second WHERE ((user_id = 6) AND ((hashint4(user_id) >= 0) AND (hashint4(user_id) <= 1073741823)))
+DEBUG:  predicate pruning for shardId 13300004
+DEBUG:  predicate pruning for shardId 13300005
+DEBUG:  predicate pruning for shardId 13300006
+DEBUG:  predicate pruning for shardId 13300007
+DEBUG:  Skipping target shard interval 13300003 since SELECT query for it pruned away
+DEBUG:  ProcessQuery
+DEBUG:  Plan is router executable
+DEBUG:  CommitTransactionCommand
+DEBUG:  CommitTransaction
+DEBUG:  name: unnamed; blockState:       STARTED; state: INPROGR, xid/subid/cid: 0/1/0, nestlvl: 1, children: 
+SET client_min_messages TO INFO;
+DEBUG:  StartTransactionCommand
+DEBUG:  StartTransaction
+DEBUG:  name: unnamed; blockState:       DEFAULT; state: INPROGR, xid/subid/cid: 0/1/0, nestlvl: 1, children: 
+DEBUG:  ProcessUtility
+-- mark the unhealthy placement as healthy again for the next tests
+UPDATE pg_dist_shard_placement SET shardstate = 1 WHERE shardid = 13300004 AND nodeport = :worker_1_port;
+-- now that we should show that it works if one of the target shard interval is not healthy
+UPDATE pg_dist_shard_placement SET shardstate = 3 WHERE shardid = 13300000 AND nodeport = :worker_1_port;
+truncate raw_events_first;
+SET client_min_messages TO DEBUG4;
+DEBUG:  CommitTransactionCommand
+DEBUG:  CommitTransaction
+DEBUG:  name: unnamed; blockState:       STARTED; state: INPROGR, xid/subid/cid: 0/1/0, nestlvl: 1, children: 
+-- this should work
+INSERT INTO raw_events_first SELECT * FROM raw_events_second;
+DEBUG:  StartTransactionCommand
+DEBUG:  StartTransaction
+DEBUG:  name: unnamed; blockState:       DEFAULT; state: INPROGR, xid/subid/cid: 0/1/0, nestlvl: 1, children: 
+DEBUG:  predicate pruning for shardId 13300005
+DEBUG:  predicate pruning for shardId 13300006
+DEBUG:  predicate pruning for shardId 13300007
+DEBUG:  distributed statement: INSERT INTO public.raw_events_first_13300000 AS citus_table_alias (user_id, "time", value_1, value_2, value_3, value_4) SELECT user_id, "time", value_1, value_2, value_3, value_4 FROM public.raw_events_second_13300004 raw_events_second WHERE ((hashint4(user_id) >= '-2147483648'::integer) AND (hashint4(user_id) <= '-1073741825'::integer))
+DEBUG:  predicate pruning for shardId 13300004
+DEBUG:  predicate pruning for shardId 13300006
+DEBUG:  predicate pruning for shardId 13300007
+DEBUG:  distributed statement: INSERT INTO public.raw_events_first_13300001 AS citus_table_alias (user_id, "time", value_1, value_2, value_3, value_4) SELECT user_id, "time", value_1, value_2, value_3, value_4 FROM public.raw_events_second_13300005 raw_events_second WHERE ((hashint4(user_id) >= '-1073741824'::integer) AND (hashint4(user_id) <= '-1'::integer))
+DEBUG:  predicate pruning for shardId 13300004
+DEBUG:  predicate pruning for shardId 13300005
+DEBUG:  predicate pruning for shardId 13300007
+DEBUG:  distributed statement: INSERT INTO public.raw_events_first_13300002 AS citus_table_alias (user_id, "time", value_1, value_2, value_3, value_4) SELECT user_id, "time", value_1, value_2, value_3, value_4 FROM public.raw_events_second_13300006 raw_events_second WHERE ((hashint4(user_id) >= 0) AND (hashint4(user_id) <= 1073741823))
+DEBUG:  predicate pruning for shardId 13300004
+DEBUG:  predicate pruning for shardId 13300005
+DEBUG:  predicate pruning for shardId 13300006
+DEBUG:  distributed statement: INSERT INTO public.raw_events_first_13300003 AS citus_table_alias (user_id, "time", value_1, value_2, value_3, value_4) SELECT user_id, "time", value_1, value_2, value_3, value_4 FROM public.raw_events_second_13300007 raw_events_second WHERE ((hashint4(user_id) >= 1073741824) AND (hashint4(user_id) <= 2147483647))
+DEBUG:  ProcessQuery
+DEBUG:  Plan is router executable
+DEBUG:  CommitTransactionCommand
+DEBUG:  CommitTransaction
+DEBUG:  name: unnamed; blockState:       STARTED; state: INPROGR, xid/subid/cid: 0/1/0, nestlvl: 1, children: 
+DEBUG:  sent COMMIT over connection 13300001
+DEBUG:  sent COMMIT over connection 13300001
+DEBUG:  sent COMMIT over connection 13300000
+DEBUG:  sent COMMIT over connection 13300002
+DEBUG:  sent COMMIT over connection 13300002
+DEBUG:  sent COMMIT over connection 13300003
+DEBUG:  sent COMMIT over connection 13300003
+SET client_min_messages TO INFO;
+DEBUG:  StartTransactionCommand
+DEBUG:  StartTransaction
+DEBUG:  name: unnamed; blockState:       DEFAULT; state: INPROGR, xid/subid/cid: 0/1/0, nestlvl: 1, children: 
+DEBUG:  ProcessUtility
+truncate raw_events_first;
+SET client_min_messages TO DEBUG4;
+DEBUG:  CommitTransactionCommand
+DEBUG:  CommitTransaction
+DEBUG:  name: unnamed; blockState:       STARTED; state: INPROGR, xid/subid/cid: 0/1/0, nestlvl: 1, children: 
+-- this should also work
+INSERT INTO raw_events_first SELECT * FROM raw_events_second WHERE user_id = 5;
+DEBUG:  StartTransactionCommand
+DEBUG:  StartTransaction
+DEBUG:  name: unnamed; blockState:       DEFAULT; state: INPROGR, xid/subid/cid: 0/1/0, nestlvl: 1, children: 
+DEBUG:  predicate pruning for shardId 13300005
+DEBUG:  predicate pruning for shardId 13300006
+DEBUG:  predicate pruning for shardId 13300007
+DEBUG:  distributed statement: INSERT INTO public.raw_events_first_13300000 AS citus_table_alias (user_id, "time", value_1, value_2, value_3, value_4) SELECT user_id, "time", value_1, value_2, value_3, value_4 FROM public.raw_events_second_13300004 raw_events_second WHERE ((user_id = 5) AND ((hashint4(user_id) >= '-2147483648'::integer) AND (hashint4(user_id) <= '-1073741825'::integer)))
+DEBUG:  predicate pruning for shardId 13300004
+DEBUG:  predicate pruning for shardId 13300005
+DEBUG:  predicate pruning for shardId 13300006
+DEBUG:  predicate pruning for shardId 13300007
+DEBUG:  Skipping target shard interval 13300001 since SELECT query for it pruned away
+DEBUG:  predicate pruning for shardId 13300004
+DEBUG:  predicate pruning for shardId 13300005
+DEBUG:  predicate pruning for shardId 13300006
+DEBUG:  predicate pruning for shardId 13300007
+DEBUG:  Skipping target shard interval 13300002 since SELECT query for it pruned away
+DEBUG:  predicate pruning for shardId 13300004
+DEBUG:  predicate pruning for shardId 13300005
+DEBUG:  predicate pruning for shardId 13300006
+DEBUG:  predicate pruning for shardId 13300007
+DEBUG:  Skipping target shard interval 13300003 since SELECT query for it pruned away
+DEBUG:  ProcessQuery
+DEBUG:  Plan is router executable
+DEBUG:  CommitTransactionCommand
+DEBUG:  CommitTransaction
+DEBUG:  name: unnamed; blockState:       STARTED; state: INPROGR, xid/subid/cid: 0/1/0, nestlvl: 1, children: 

src/test/regress/sql/multi_insert_select.sql

@@ -606,3 +606,54 @@ SET client_min_messages TO INFO;
 -- Views does not work
 CREATE VIEW test_view AS SELECT * FROM raw_events_first;
 INSERT INTO raw_events_second SELECT * FROM test_view;
+
+-- we need this in our next test
+truncate raw_events_first;
+
+SET client_min_messages TO DEBUG4;
+
+-- first show that the query works now
+INSERT INTO raw_events_first SELECT * FROM raw_events_second;
+
+SET client_min_messages TO INFO;
+truncate raw_events_first;
+SET client_min_messages TO DEBUG4;
+
+-- now show that it works for a single shard query as well
+INSERT INTO raw_events_first SELECT * FROM raw_events_second WHERE user_id = 5;
+
+SET client_min_messages TO INFO;
+
+-- if a single shard of the SELECT is unhealty, the query should fail
+UPDATE pg_dist_shard_placement SET shardstate = 3 WHERE shardid = 13300004 AND nodeport = :worker_1_port;
+truncate raw_events_first;
+SET client_min_messages TO DEBUG4;
+
+-- this should fail
+INSERT INTO raw_events_first SELECT * FROM raw_events_second;
+
+-- this should also fail
+INSERT INTO raw_events_first SELECT * FROM raw_events_second WHERE user_id = 5;
+
+-- but this should work given that it hits different shard
+INSERT INTO raw_events_first SELECT * FROM raw_events_second WHERE user_id = 6;
+
+SET client_min_messages TO INFO;
+
+-- mark the unhealthy placement as healthy again for the next tests
+UPDATE pg_dist_shard_placement SET shardstate = 1 WHERE shardid = 13300004 AND nodeport = :worker_1_port;
+
+-- now that we should show that it works if one of the target shard interval is not healthy
+UPDATE pg_dist_shard_placement SET shardstate = 3 WHERE shardid = 13300000 AND nodeport = :worker_1_port;
+truncate raw_events_first;
+SET client_min_messages TO DEBUG4;
+
+-- this should work
+INSERT INTO raw_events_first SELECT * FROM raw_events_second;
+
+SET client_min_messages TO INFO;
+truncate raw_events_first;
+SET client_min_messages TO DEBUG4;
+
+-- this should also work
+INSERT INTO raw_events_first SELECT * FROM raw_events_second WHERE user_id = 5;