Disable citus.enable_non_colocated_router_query_pushdown by default (#6909)

Fixes #6779.

DESCRIPTION: Disables citus.enable_non_colocated_router_query_pushdown
GUC by default to ensure generating a consistent distributed plan for
the queries that reference non-colocated distributed tables

We already have tests for the cases where this GUC is disabled,
so I'm not adding any more tests in this PR.

Also make multi_insert_select_window idempotent.

Related to: #6793

src/backend/distributed/shared_library_init.c

@@ -1346,7 +1346,7 @@ RegisterCitusConfigVariables(void)
 					 "or altering the shard count of one of those distributed "
 					 "tables."),
 		&EnableNonColocatedRouterQueryPushdown,
-		true,
+		false,
 		PGC_USERSET,
 		GUC_NO_SHOW_ALL,
 		NULL, NULL, NULL);

src/test/regress/expected/insert_select_repartition.out

@@ -756,6 +756,7 @@ DEALLOCATE insert_plan;
 --
 TRUNCATE target_table;
 SET client_min_messages TO DEBUG2;
+SET citus.enable_non_colocated_router_query_pushdown TO ON;
 WITH r AS (
   INSERT INTO target_table SELECT * FROM source_table RETURNING *
 )
@@ -777,6 +778,7 @@ DEBUG:  partitioning SELECT query by column index 0 with name 'a'
 DEBUG:  distributed statement: INSERT INTO insert_select_repartition.target_table_4213610 AS citus_table_alias (a, b) SELECT intermediate_result.a, intermediate_result.b FROM read_intermediate_results('{repartitioned_results_xxxxx_from_4213606_to_0,repartitioned_results_xxxxx_from_4213607_to_0}'::text[], 'binary'::citus_copy_format) intermediate_result(a integer, b integer)
 DEBUG:  distributed statement: INSERT INTO insert_select_repartition.target_table_4213611 AS citus_table_alias (a, b) SELECT intermediate_result.a, intermediate_result.b FROM read_intermediate_results('{repartitioned_results_xxxxx_from_4213607_to_1}'::text[], 'binary'::citus_copy_format) intermediate_result(a integer, b integer)
 DEBUG:  distributed statement: INSERT INTO insert_select_repartition.target_table_4213612 AS citus_table_alias (a, b) SELECT intermediate_result.a, intermediate_result.b FROM read_intermediate_results('{repartitioned_results_xxxxx_from_4213609_to_2}'::text[], 'binary'::citus_copy_format) intermediate_result(a integer, b integer)
+RESET citus.enable_non_colocated_router_query_pushdown;
 RESET client_min_messages;
 SELECT * FROM target_table ORDER BY a, b;
  a |  b

src/test/regress/expected/insert_select_repartition_0.out

@@ -756,6 +756,7 @@ DEALLOCATE insert_plan;
 --
 TRUNCATE target_table;
 SET client_min_messages TO DEBUG2;
+SET citus.enable_non_colocated_router_query_pushdown TO ON;
 WITH r AS (
   INSERT INTO target_table SELECT * FROM source_table RETURNING *
 )
@@ -777,6 +778,7 @@ DEBUG:  partitioning SELECT query by column index 0 with name 'a'
 DEBUG:  distributed statement: INSERT INTO insert_select_repartition.target_table_4213610 AS citus_table_alias (a, b) SELECT a, b FROM read_intermediate_results('{repartitioned_results_xxxxx_from_4213606_to_0,repartitioned_results_xxxxx_from_4213607_to_0}'::text[], 'binary'::citus_copy_format) intermediate_result(a integer, b integer)
 DEBUG:  distributed statement: INSERT INTO insert_select_repartition.target_table_4213611 AS citus_table_alias (a, b) SELECT a, b FROM read_intermediate_results('{repartitioned_results_xxxxx_from_4213607_to_1}'::text[], 'binary'::citus_copy_format) intermediate_result(a integer, b integer)
 DEBUG:  distributed statement: INSERT INTO insert_select_repartition.target_table_4213612 AS citus_table_alias (a, b) SELECT a, b FROM read_intermediate_results('{repartitioned_results_xxxxx_from_4213609_to_2}'::text[], 'binary'::citus_copy_format) intermediate_result(a integer, b integer)
+RESET citus.enable_non_colocated_router_query_pushdown;
 RESET client_min_messages;
 SELECT * FROM target_table ORDER BY a, b;
  a |  b

src/test/regress/expected/insert_select_single_shard_table.out

@@ -160,34 +160,33 @@ DEBUG:  Collecting INSERT ... SELECT results on coordinator
 INSERT INTO distributed_table_c1_t1 SELECT nullkey_c1_t2.a, nullkey_c1_t2.b FROM nullkey_c1_t2 LEFT JOIN nullkey_c2_t1 USING (a);
 DEBUG:  distributed INSERT ... SELECT cannot reference a distributed table without a shard key together with non-colocated distributed tables
 ERROR:  queries that reference a distributed table without a shard key can only reference colocated distributed tables or reference tables
-DETAIL:  found no worker with all shard placements
+DETAIL:  router planner does not support queries that reference non-colocated distributed tables
 INSERT INTO distributed_table_c1_t1 SELECT * FROM nullkey_c1_t1 UNION SELECT * FROM nullkey_c2_t1;
 DEBUG:  Set operations are not allowed in distributed INSERT ... SELECT queries
 ERROR:  queries that reference a distributed table without a shard key can only reference colocated distributed tables or reference tables
-DETAIL:  found no worker with all shard placements
+DETAIL:  router planner does not support queries that reference non-colocated distributed tables
 -- use a distributed table that is colocated with the target table
 INSERT INTO distributed_table_c1_t1 SELECT nullkey_c1_t1.a, nullkey_c1_t1.b FROM nullkey_c1_t1 JOIN distributed_table_c1_t2 USING (a);
 DEBUG:  distributed INSERT ... SELECT cannot reference a distributed table without a shard key together with non-colocated distributed tables
 ERROR:  queries that reference a distributed table without a shard key can only reference colocated distributed tables or reference tables
-DETAIL:  Router planner cannot handle multi-shard select queries
+DETAIL:  router planner does not support queries that reference non-colocated distributed tables
 INSERT INTO distributed_table_c1_t1 SELECT distributed_table_c1_t2.a, distributed_table_c1_t2.b FROM nullkey_c1_t1 JOIN distributed_table_c1_t2 USING (a);
 DEBUG:  distributed INSERT ... SELECT cannot reference a distributed table without a shard key together with non-colocated distributed tables
 ERROR:  queries that reference a distributed table without a shard key can only reference colocated distributed tables or reference tables
-DETAIL:  Router planner cannot handle multi-shard select queries
+DETAIL:  router planner does not support queries that reference non-colocated distributed tables
 INSERT INTO distributed_table_c1_t1 SELECT distributed_table_c1_t2.a, distributed_table_c1_t2.b FROM nullkey_c1_t1 JOIN distributed_table_c1_t2 USING (b);
 DEBUG:  distributed INSERT ... SELECT cannot reference a distributed table without a shard key together with non-colocated distributed tables
 ERROR:  queries that reference a distributed table without a shard key can only reference colocated distributed tables or reference tables
-DETAIL:  Router planner cannot handle multi-shard select queries
+DETAIL:  router planner does not support queries that reference non-colocated distributed tables
 INSERT INTO distributed_table_c1_t1 SELECT distributed_table_c1_t2.a, distributed_table_c1_t2.b FROM nullkey_c1_t1 JOIN distributed_table_c1_t2 USING (a) WHERE distributed_table_c1_t2.a = 1;
 DEBUG:  distributed INSERT ... SELECT cannot reference a distributed table without a shard key together with non-colocated distributed tables
-DEBUG:  Creating router plan
-DEBUG:  query has a single distribution column value: 1
-DEBUG:  Collecting INSERT ... SELECT results on coordinator
+ERROR:  queries that reference a distributed table without a shard key can only reference colocated distributed tables or reference tables
+DETAIL:  router planner does not support queries that reference non-colocated distributed tables
 -- use a distributed table that is not colocated with the target table
 INSERT INTO distributed_table_c1_t1 SELECT nullkey_c1_t2.a, nullkey_c1_t2.b FROM nullkey_c1_t2 JOIN distributed_table_c2_t1 USING (a);
 DEBUG:  distributed INSERT ... SELECT cannot reference a distributed table without a shard key together with non-colocated distributed tables
 ERROR:  queries that reference a distributed table without a shard key can only reference colocated distributed tables or reference tables
-DETAIL:  Router planner cannot handle multi-shard select queries
+DETAIL:  router planner does not support queries that reference non-colocated distributed tables
 -- use a citus local table
 INSERT INTO distributed_table_c1_t1 SELECT nullkey_c1_t1.a, nullkey_c1_t1.b FROM nullkey_c1_t1 JOIN citus_local_table USING (a);
 DEBUG:  distributed INSERT ... SELECT cannot select from distributed tables and local tables at the same time
@@ -257,25 +256,24 @@ DEBUG:  Collecting INSERT ... SELECT results on coordinator
 INSERT INTO reference_table SELECT nullkey_c1_t2.a, nullkey_c1_t2.b FROM nullkey_c1_t2 LEFT JOIN nullkey_c2_t1 USING (a);
 DEBUG:  only reference tables may be queried when targeting a reference table with distributed INSERT ... SELECT
 ERROR:  queries that reference a distributed table without a shard key can only reference colocated distributed tables or reference tables
-DETAIL:  found no worker with all shard placements
+DETAIL:  router planner does not support queries that reference non-colocated distributed tables
 -- use a distributed table
 INSERT INTO reference_table SELECT nullkey_c1_t1.a, nullkey_c1_t1.b FROM nullkey_c1_t1 JOIN distributed_table_c1_t2 USING (a);
 DEBUG:  only reference tables may be queried when targeting a reference table with distributed INSERT ... SELECT
 ERROR:  queries that reference a distributed table without a shard key can only reference colocated distributed tables or reference tables
-DETAIL:  Router planner cannot handle multi-shard select queries
+DETAIL:  router planner does not support queries that reference non-colocated distributed tables
 INSERT INTO reference_table SELECT distributed_table_c1_t2.a, distributed_table_c1_t2.b FROM nullkey_c1_t1 JOIN distributed_table_c1_t2 USING (a);
 DEBUG:  only reference tables may be queried when targeting a reference table with distributed INSERT ... SELECT
 ERROR:  queries that reference a distributed table without a shard key can only reference colocated distributed tables or reference tables
-DETAIL:  Router planner cannot handle multi-shard select queries
+DETAIL:  router planner does not support queries that reference non-colocated distributed tables
 INSERT INTO reference_table SELECT distributed_table_c1_t2.a, distributed_table_c1_t2.b FROM nullkey_c1_t1 JOIN distributed_table_c1_t2 USING (b);
 DEBUG:  only reference tables may be queried when targeting a reference table with distributed INSERT ... SELECT
 ERROR:  queries that reference a distributed table without a shard key can only reference colocated distributed tables or reference tables
-DETAIL:  Router planner cannot handle multi-shard select queries
+DETAIL:  router planner does not support queries that reference non-colocated distributed tables
 INSERT INTO reference_table SELECT distributed_table_c1_t2.a, distributed_table_c1_t2.b FROM nullkey_c1_t1 JOIN distributed_table_c1_t2 USING (a) WHERE distributed_table_c1_t2.a = 1;
 DEBUG:  only reference tables may be queried when targeting a reference table with distributed INSERT ... SELECT
-DEBUG:  Creating router plan
-DEBUG:  query has a single distribution column value: 1
-DEBUG:  Collecting INSERT ... SELECT results on coordinator
+ERROR:  queries that reference a distributed table without a shard key can only reference colocated distributed tables or reference tables
+DETAIL:  router planner does not support queries that reference non-colocated distributed tables
 -- use a citus local table
 INSERT INTO reference_table SELECT nullkey_c1_t1.a, nullkey_c1_t1.b FROM nullkey_c1_t1 JOIN citus_local_table USING (a);
 DEBUG:  distributed INSERT ... SELECT cannot select from distributed tables and local tables at the same time
@@ -320,7 +318,7 @@ DEBUG:  Collecting INSERT ... SELECT results on coordinator
 INSERT INTO citus_local_table SELECT nullkey_c1_t1.a, nullkey_c1_t1.b FROM nullkey_c1_t1 JOIN distributed_table_c1_t2 USING (a);
 DEBUG:  distributed INSERT ... SELECT cannot insert into a local table that is added to metadata
 ERROR:  queries that reference a distributed table without a shard key can only reference colocated distributed tables or reference tables
-DETAIL:  Router planner cannot handle multi-shard select queries
+DETAIL:  router planner does not support queries that reference non-colocated distributed tables
 -- use a citus local table
 INSERT INTO citus_local_table SELECT nullkey_c1_t1.a, nullkey_c1_t1.b FROM nullkey_c1_t1 JOIN citus_local_table USING (a);
 DEBUG:  distributed INSERT ... SELECT cannot select from distributed tables and local tables at the same time
@@ -383,7 +381,7 @@ DEBUG:  Collecting INSERT ... SELECT results on coordinator
 INSERT INTO nullkey_c1_t1 SELECT distributed_table_c1_t2.a, distributed_table_c1_t2.b FROM distributed_table_c1_t2 JOIN nullkey_c1_t1 USING (a);
 DEBUG:  distributed INSERT ... SELECT cannot reference a distributed table without a shard key together with non-colocated distributed tables
 ERROR:  queries that reference a distributed table without a shard key can only reference colocated distributed tables or reference tables
-DETAIL:  Router planner cannot handle multi-shard select queries
+DETAIL:  router planner does not support queries that reference non-colocated distributed tables
 -- use a non-colocated single-shard table
 INSERT INTO nullkey_c2_t1 SELECT q.* FROM (SELECT reference_table.* FROM reference_table LEFT JOIN nullkey_c1_t1 USING (a)) q JOIN nullkey_c1_t2 USING (a);
 DEBUG:  distributed INSERT ... SELECT cannot reference a distributed table without a shard key together with non-colocated distributed tables
@@ -516,7 +514,7 @@ CROSS JOIN (
 ) t2;
 DEBUG:  distributed INSERT ... SELECT cannot insert into a local table that is added to metadata
 ERROR:  queries that reference a distributed table without a shard key can only reference colocated distributed tables or reference tables
-DETAIL:  found no worker with all shard placements
+DETAIL:  router planner does not support queries that reference non-colocated distributed tables
 INSERT INTO distributed_table_c1_t1 (a, b)
 SELECT t1.a, t2.b
 FROM reference_table t1
@@ -541,7 +539,7 @@ JOIN (
 WHERE t2.rn > 2;
 DEBUG:  Window functions without PARTITION BY on distribution column is currently unsupported
 ERROR:  queries that reference a distributed table without a shard key can only reference colocated distributed tables or reference tables
-DETAIL:  Router planner cannot handle multi-shard select queries
+DETAIL:  router planner does not support queries that reference non-colocated distributed tables
 INSERT INTO distributed_table_c1_t1 (a, b)
 SELECT t1.a, t2.b
 FROM nullkey_c1_t1 t1

src/test/regress/expected/multi_insert_select_window.out

@@ -816,4 +816,5 @@ FROM (
             user_id
         )
 ) AS ftop;
+TRUNCATE agg_results_window;
 DROP VIEW view_with_window_func;

src/test/regress/expected/multi_router_planner.out

@@ -830,7 +830,7 @@ SELECT a.author_id as first_author, b.word_count as second_word_count
 	FROM articles_hash a, single_shard b
 	WHERE a.author_id = 2 and a.author_id = b.author_id
 	LIMIT 3;
-DEBUG:  found no worker with all shard placements
+DEBUG:  router planner does not support queries that reference non-colocated distributed tables
 DEBUG:  generating subplan XXX_1 for CTE single_shard: SELECT id, author_id, title, word_count FROM multi_router_planner.articles_single_shard_hash
 DEBUG:  Creating router plan
 DEBUG:  Plan XXX query after replacing subqueries and CTEs: SELECT a.author_id AS first_author, b.word_count AS second_word_count FROM multi_router_planner.articles_hash a, (SELECT intermediate_result.id, intermediate_result.author_id, intermediate_result.title, intermediate_result.word_count FROM read_intermediate_result('XXX_1'::text, 'binary'::citus_copy_format) intermediate_result(id bigint, author_id bigint, title character varying(20), word_count integer)) b WHERE ((a.author_id OPERATOR(pg_catalog.=) 2) AND (a.author_id OPERATOR(pg_catalog.=) b.author_id)) LIMIT 3
@@ -1513,21 +1513,41 @@ DEBUG:  Creating router plan
 SELECT a.author_id as first_author, b.word_count as second_word_count
 	FROM articles_hash a, articles_single_shard_hash b
 	WHERE a.author_id = 10 and a.author_id = b.author_id and int4eq(1, 1);
-DEBUG:  Creating router plan
-DEBUG:  query has a single distribution column value: 10
- first_author | second_word_count
----------------------------------------------------------------------
-           10 |             19519
-           10 |             19519
-           10 |             19519
-           10 |             19519
-           10 |             19519
-(5 rows)
-
+DEBUG:  router planner does not support queries that reference non-colocated distributed tables
+DEBUG:  join prunable for task partitionId 0 and 1
+DEBUG:  join prunable for task partitionId 0 and 2
+DEBUG:  join prunable for task partitionId 0 and 3
+DEBUG:  join prunable for task partitionId 1 and 0
+DEBUG:  join prunable for task partitionId 1 and 2
+DEBUG:  join prunable for task partitionId 1 and 3
+DEBUG:  join prunable for task partitionId 2 and 0
+DEBUG:  join prunable for task partitionId 2 and 1
+DEBUG:  join prunable for task partitionId 2 and 3
+DEBUG:  join prunable for task partitionId 3 and 0
+DEBUG:  join prunable for task partitionId 3 and 1
+DEBUG:  join prunable for task partitionId 3 and 2
+DEBUG:  pruning merge fetch taskId 1
+DETAIL:  Creating dependency on merge taskId 2
+DEBUG:  pruning merge fetch taskId 2
+DETAIL:  Creating dependency on merge taskId 2
+DEBUG:  pruning merge fetch taskId 4
+DETAIL:  Creating dependency on merge taskId 4
+DEBUG:  pruning merge fetch taskId 5
+DETAIL:  Creating dependency on merge taskId 4
+DEBUG:  pruning merge fetch taskId 7
+DETAIL:  Creating dependency on merge taskId 6
+DEBUG:  pruning merge fetch taskId 8
+DETAIL:  Creating dependency on merge taskId 6
+DEBUG:  pruning merge fetch taskId 10
+DETAIL:  Creating dependency on merge taskId 8
+DEBUG:  pruning merge fetch taskId 11
+DETAIL:  Creating dependency on merge taskId 8
+ERROR:  the query contains a join that requires repartitioning
+HINT:  Set citus.enable_repartition_joins to on to enable repartitioning
 SELECT a.author_id as first_author, b.word_count as second_word_count
 	FROM articles_hash a, articles_single_shard_hash b
 	WHERE a.author_id = 10 and a.author_id = b.author_id and int4eq(1, 2);
-DEBUG:  Creating router plan
+DEBUG:  router planner does not support queries that reference non-colocated distributed tables
  first_author | second_word_count
 ---------------------------------------------------------------------
 (0 rows)
@@ -1565,7 +1585,7 @@ SELECT a.author_id as first_author, b.word_count as second_word_count
 	FROM articles_hash a, articles_single_shard_hash b
 	WHERE a.author_id = 10 and a.author_id = b.author_id and
 		date_ne_timestamp('1954-04-11', '1954-04-11'::timestamp);
-DEBUG:  Creating router plan
+DEBUG:  router planner does not support queries that reference non-colocated distributed tables
  first_author | second_word_count
 ---------------------------------------------------------------------
 (0 rows)
@@ -2008,7 +2028,7 @@ RESET citus.enable_non_colocated_router_query_pushdown;
 -- not router plannable
 SELECT * FROM articles_hash ar join authors_range au on (ar.author_id = au.id)
 	WHERE ar.author_id = 3;
-DEBUG:  found no worker with all shard placements
+DEBUG:  router planner does not support queries that reference non-colocated distributed tables
 DEBUG:  join prunable for task partitionId 0 and 1
 DEBUG:  join prunable for task partitionId 0 and 2
 DEBUG:  join prunable for task partitionId 0 and 3

src/test/regress/expected/query_single_shard_table.out

@@ -233,7 +233,7 @@ DEBUG:  Creating router plan
 --    with other table types
 SELECT COUNT(*) FROM distributed_table d1, nullkey_c1_t1;
 ERROR:  queries that reference a distributed table without a shard key can only reference colocated distributed tables or reference tables
-DETAIL:  Router planner cannot handle multi-shard select queries
+DETAIL:  router planner does not support queries that reference non-colocated distributed tables
 SELECT COUNT(*) FROM reference_table d1, nullkey_c1_t1;
 DEBUG:  Creating router plan
  count
@@ -258,7 +258,7 @@ DEBUG:  Creating router plan
 --    with a non-colocated single-shard table
 SELECT COUNT(*) FROM nullkey_c1_t1 d1, nullkey_c2_t1;
 ERROR:  queries that reference a distributed table without a shard key can only reference colocated distributed tables or reference tables
-DETAIL:  found no worker with all shard placements
+DETAIL:  router planner does not support queries that reference non-colocated distributed tables
 -- First, show that nullkey_c1_t1 and nullkey_c3_t1 are not colocated.
 SELECT
   (SELECT colocationid FROM pg_dist_partition WHERE logicalrelid = 'query_single_shard_table.nullkey_c1_t1'::regclass) !=
@@ -359,38 +359,38 @@ DEBUG:  Creating router plan
 -- non-colocated inner joins between single-shard tables
 SELECT * FROM nullkey_c1_t1 JOIN nullkey_c2_t1 USING(a) ORDER BY 1,2,3;
 ERROR:  queries that reference a distributed table without a shard key can only reference colocated distributed tables or reference tables
-DETAIL:  found no worker with all shard placements
+DETAIL:  router planner does not support queries that reference non-colocated distributed tables
 SELECT COUNT(*) FROM nullkey_c1_t1 t1
 JOIN LATERAL (
     SELECT * FROM nullkey_c2_t2 t2 WHERE t2.b > t1.a
 ) q USING(a);
 ERROR:  queries that reference a distributed table without a shard key can only reference colocated distributed tables or reference tables
-DETAIL:  found no worker with all shard placements
+DETAIL:  router planner does not support queries that reference non-colocated distributed tables
 -- non-colocated outer joins between single-shard tables
 SELECT * FROM nullkey_c1_t1 LEFT JOIN nullkey_c2_t2 USING(a) ORDER BY 1,2,3 LIMIT 4;
 ERROR:  queries that reference a distributed table without a shard key can only reference colocated distributed tables or reference tables
-DETAIL:  found no worker with all shard placements
+DETAIL:  router planner does not support queries that reference non-colocated distributed tables
 SELECT * FROM nullkey_c1_t1 FULL JOIN nullkey_c2_t2 USING(a) ORDER BY 1,2,3 LIMIT 4;
 ERROR:  queries that reference a distributed table without a shard key can only reference colocated distributed tables or reference tables
-DETAIL:  found no worker with all shard placements
+DETAIL:  router planner does not support queries that reference non-colocated distributed tables
 SELECT * FROM nullkey_c1_t1 t1
 LEFT JOIN LATERAL (
     SELECT * FROM nullkey_c2_t2 t2 WHERE t2.b > t1.a
 ) q USING(a) ORDER BY 1,2,3 OFFSET 3 LIMIT 4;
 ERROR:  queries that reference a distributed table without a shard key can only reference colocated distributed tables or reference tables
-DETAIL:  found no worker with all shard placements
+DETAIL:  router planner does not support queries that reference non-colocated distributed tables
 SELECT COUNT(*) FROM nullkey_c1_t1 t1
 LEFT JOIN LATERAL (
     SELECT * FROM nullkey_c2_t2 t2 WHERE t2.b > t1.a
 ) q USING(a);
 ERROR:  queries that reference a distributed table without a shard key can only reference colocated distributed tables or reference tables
-DETAIL:  found no worker with all shard placements
+DETAIL:  router planner does not support queries that reference non-colocated distributed tables
 SELECT COUNT(*) FROM nullkey_c1_t1 t1
 WHERE EXISTS (
     SELECT * FROM nullkey_c2_t2 t2 WHERE t2.b > t1.a
 );
 ERROR:  queries that reference a distributed table without a shard key can only reference colocated distributed tables or reference tables
-DETAIL:  found no worker with all shard placements
+DETAIL:  router planner does not support queries that reference non-colocated distributed tables
 SELECT COUNT(*) FROM nullkey_c1_t1 t1
 WHERE t1.b IN (
     SELECT b+1 FROM nullkey_c2_t2 t2 WHERE t2.b = t1.a
@@ -430,19 +430,19 @@ DETAIL:  Local tables cannot be used in distributed queries.
 -- join with a distributed table
 SELECT * FROM distributed_table d1 JOIN nullkey_c1_t1 USING(a);
 ERROR:  queries that reference a distributed table without a shard key can only reference colocated distributed tables or reference tables
-DETAIL:  Router planner cannot handle multi-shard select queries
+DETAIL:  router planner does not support queries that reference non-colocated distributed tables
 SELECT COUNT(*) FROM nullkey_c1_t1 t1
 JOIN LATERAL (
     SELECT * FROM distributed_table t2 WHERE t2.b > t1.a
 ) q USING(a);
 ERROR:  queries that reference a distributed table without a shard key can only reference colocated distributed tables or reference tables
-DETAIL:  Router planner cannot handle multi-shard select queries
+DETAIL:  router planner does not support queries that reference non-colocated distributed tables
 SELECT COUNT(*) FROM distributed_table t1
 JOIN LATERAL (
     SELECT * FROM nullkey_c1_t1 t2 WHERE t2.b > t1.a
 ) q USING(a);
 ERROR:  queries that reference a distributed table without a shard key can only reference colocated distributed tables or reference tables
-DETAIL:  Router planner cannot handle multi-shard select queries
+DETAIL:  router planner does not support queries that reference non-colocated distributed tables
 -- outer joins with different table types
 SELECT COUNT(*) FROM nullkey_c1_t1 LEFT JOIN reference_table USING(a);
 DEBUG:  Creating router plan
@@ -488,7 +488,7 @@ ERROR:  queries that reference a distributed table without a shard key can only
 DETAIL:  Router planner does not support append-partitioned tables.
 SELECT COUNT(*) FROM nullkey_c1_t1 JOIN range_table USING(a);
 ERROR:  queries that reference a distributed table without a shard key can only reference colocated distributed tables or reference tables
-DETAIL:  Router planner cannot handle multi-shard select queries
+DETAIL:  router planner does not support queries that reference non-colocated distributed tables
 SET citus.enable_non_colocated_router_query_pushdown TO ON;
 SELECT COUNT(*) FROM nullkey_c1_t1 JOIN range_table USING(a) WHERE range_table.a = 20;
 DEBUG:  Creating router plan
@@ -620,19 +620,19 @@ LEFT JOIN LATERAL (
     SELECT * FROM distributed_table t2 WHERE t2.b > t1.a
 ) q USING(a);
 ERROR:  queries that reference a distributed table without a shard key can only reference colocated distributed tables or reference tables
-DETAIL:  Router planner cannot handle multi-shard select queries
+DETAIL:  router planner does not support queries that reference non-colocated distributed tables
 SELECT COUNT(*) FROM nullkey_c1_t1 t1
 WHERE EXISTS (
     SELECT * FROM distributed_table t2 WHERE t2.b > t1.a
 );
 ERROR:  queries that reference a distributed table without a shard key can only reference colocated distributed tables or reference tables
-DETAIL:  Router planner cannot handle multi-shard select queries
+DETAIL:  router planner does not support queries that reference non-colocated distributed tables
 SELECT COUNT(*) FROM nullkey_c1_t1 t1
 WHERE NOT EXISTS (
     SELECT * FROM distributed_table t2 WHERE t2.b > t1.a
 );
 ERROR:  queries that reference a distributed table without a shard key can only reference colocated distributed tables or reference tables
-DETAIL:  Router planner cannot handle multi-shard select queries
+DETAIL:  router planner does not support queries that reference non-colocated distributed tables
 SELECT COUNT(*) FROM nullkey_c1_t1 t1
 WHERE t1.b IN (
     SELECT b+1 FROM distributed_table t2 WHERE t2.b = t1.a
@@ -650,13 +650,13 @@ LEFT JOIN LATERAL (
     SELECT * FROM nullkey_c1_t1 t2 WHERE t2.b > t1.a
 ) q USING(a);
 ERROR:  queries that reference a distributed table without a shard key can only reference colocated distributed tables or reference tables
-DETAIL:  Router planner cannot handle multi-shard select queries
+DETAIL:  router planner does not support queries that reference non-colocated distributed tables
 SELECT COUNT(*) FROM distributed_table t1
 WHERE EXISTS (
     SELECT * FROM nullkey_c1_t1 t2 WHERE t2.b > t1.a
 );
 ERROR:  queries that reference a distributed table without a shard key can only reference colocated distributed tables or reference tables
-DETAIL:  Router planner cannot handle multi-shard select queries
+DETAIL:  router planner does not support queries that reference non-colocated distributed tables
 SELECT COUNT(*) FROM distributed_table t1
 WHERE t1.b IN (
     SELECT b+1 FROM nullkey_c1_t1 t2 WHERE t2.b = t1.a
@@ -934,7 +934,7 @@ SELECT COUNT(*) FROM level_0;
 DEBUG:  CTE level_0 is going to be inlined via distributed planning
 DEBUG:  CTE level_1 is going to be inlined via distributed planning
 ERROR:  queries that reference a distributed table without a shard key can only reference colocated distributed tables or reference tables
-DETAIL:  Router planner cannot handle multi-shard select queries
+DETAIL:  router planner does not support queries that reference non-colocated distributed tables
 -- grouping set
 SELECT
 	id, substring(title, 2, 1) AS subtitle, count(*)
@@ -1722,13 +1722,13 @@ WITH cte AS (
 )
 SELECT * FROM nullkey_c2_t1 WHERE a IN (SELECT a FROM cte);
 ERROR:  queries that reference a distributed table without a shard key can only reference colocated distributed tables or reference tables
-DETAIL:  found no worker with all shard placements
+DETAIL:  router planner does not support queries that reference non-colocated distributed tables
 WITH cte AS (
     DELETE FROM nullkey_c1_t1 WHERE a = 1 RETURNING *
 )
 SELECT * FROM distributed_table WHERE a IN (SELECT a FROM cte);
 ERROR:  queries that reference a distributed table without a shard key can only reference colocated distributed tables or reference tables
-DETAIL:  Router planner cannot handle multi-shard select queries
+DETAIL:  router planner does not support queries that reference non-colocated distributed tables
 -- Below two queries fail very late when
 -- citus.enable_non_colocated_router_query_pushdown is set to on.
 SET citus.enable_non_colocated_router_query_pushdown TO ON;

src/test/regress/expected/set_operations.out

@@ -1067,7 +1067,7 @@ DEBUG:  Creating router plan
 
 -- queries on non-colocated tables that would push down if they were not colocated are recursivelu planned
 SELECT * FROM (SELECT * FROM test UNION SELECT * FROM test_not_colocated) u ORDER BY 1,2;
-DEBUG:  Router planner cannot handle multi-shard select queries
+DEBUG:  router planner does not support queries that reference non-colocated distributed tables
 DEBUG:  Router planner cannot handle multi-shard select queries
 DEBUG:  generating subplan XXX_1 for subquery SELECT x, y FROM recursive_union.test
 DEBUG:  Router planner cannot handle multi-shard select queries
@@ -1083,7 +1083,7 @@ DEBUG:  Creating router plan
 (2 rows)
 
 SELECT * FROM (SELECT * FROM test UNION ALL SELECT * FROM test_not_colocated) u ORDER BY 1,2;
-DEBUG:  Router planner cannot handle multi-shard select queries
+DEBUG:  router planner does not support queries that reference non-colocated distributed tables
 DEBUG:  Router planner cannot handle multi-shard select queries
 DEBUG:  generating subplan XXX_1 for subquery SELECT x, y FROM recursive_union.test
 DEBUG:  Router planner cannot handle multi-shard select queries

src/test/regress/sql/insert_select_repartition.sql

@@ -389,10 +389,12 @@ DEALLOCATE insert_plan;
 TRUNCATE target_table;
 
 SET client_min_messages TO DEBUG2;
+SET citus.enable_non_colocated_router_query_pushdown TO ON;
 WITH r AS (
   INSERT INTO target_table SELECT * FROM source_table RETURNING *
 )
 INSERT INTO target_table SELECT source_table.a, max(source_table.b) FROM source_table NATURAL JOIN r GROUP BY source_table.a;
+RESET citus.enable_non_colocated_router_query_pushdown;
 RESET client_min_messages;
 
 SELECT * FROM target_table ORDER BY a, b;

src/test/regress/sql/multi_insert_select_window.sql

@@ -751,4 +751,6 @@ FROM (
         )
 ) AS ftop;
 
+TRUNCATE agg_results_window;
+
 DROP VIEW view_with_window_func;