mirror of https://github.com/citusdata/citus.git
Add alternative test outputs for change in Insert Select display pt3
citus_local_tables_queries.sql cte_inline.sql insert_select_repartition.sql intermediate_result_pruning.sql local_shard_execution.sql local_shard_execution_replicated.sql multi_mx_insert_select_repartition.sql mx_coordinator_shouldhaveshards.sql Relevant PG commit: a8d8445a7b2f80f6d0bfe97b19f90bd2cbef8759naisila/failure_pg15
naisila
parent
3c9eea031e
commit
e579e6f455
|
|
@ -1,3 +1,12 @@
|
|||
--
|
||||
-- CITUS_LOCAL_TABLES_QUERIES
|
||||
--
|
||||
-- Test queries on a distributed table with shards on the coordinator
|
||||
--
|
||||
-- This test file has an alternative output because of the change in the
|
||||
-- display of SQL-standard function's arguments in INSERT/SELECT in PG15.
|
||||
-- The alternative output can be deleted when we drop support for PG14
|
||||
--
|
||||
\set VERBOSITY terse
|
||||
SET citus.next_shard_id TO 1509000;
|
||||
SET citus.shard_replication_factor TO 1;
|
||||
|
|
@ -570,7 +579,7 @@ SELECT clear_and_init_test_tables();
|
|||
|
||||
INSERT INTO citus_local_table
|
||||
SELECT * from reference_table;
|
||||
NOTICE: executing the command locally: INSERT INTO citus_local_table_queries.citus_local_table_1509001 AS citus_table_alias (a, b) SELECT a, b FROM citus_local_table_queries.reference_table_1509003 reference_table
|
||||
NOTICE: executing the command locally: INSERT INTO citus_local_table_queries.citus_local_table_1509001 AS citus_table_alias (a, b) SELECT reference_table.a, reference_table.b FROM citus_local_table_queries.reference_table_1509003 reference_table
|
||||
INSERT INTO reference_table
|
||||
SELECT * from citus_local_table;
|
||||
NOTICE: executing the command locally: SELECT a, b FROM citus_local_table_queries.citus_local_table_1509001 citus_local_table
|
||||
|
|
@ -583,7 +592,7 @@ SELECT * from citus_local_table;
|
|||
NOTICE: executing the command locally: SELECT a, b FROM citus_local_table_queries.citus_local_table_1509001 citus_local_table
|
||||
INSERT INTO citus_local_table
|
||||
SELECT * from citus_local_table_2;
|
||||
NOTICE: executing the command locally: INSERT INTO citus_local_table_queries.citus_local_table_1509001 AS citus_table_alias (a, b) SELECT a, b FROM citus_local_table_queries.citus_local_table_2_1509002 citus_local_table_2
|
||||
NOTICE: executing the command locally: INSERT INTO citus_local_table_queries.citus_local_table_1509001 AS citus_table_alias (a, b) SELECT citus_local_table_2.a, citus_local_table_2.b FROM citus_local_table_queries.citus_local_table_2_1509002 citus_local_table_2
|
||||
INSERT INTO citus_local_table
|
||||
SELECT * from citus_local_table_2
|
||||
ORDER BY 1,2
|
||||
|
|
|
|||
File diff suppressed because it is too large
Load Diff
|
|
@ -1,3 +1,12 @@
|
|||
--
|
||||
-- CTE_INLINE
|
||||
--
|
||||
-- Test queries on a distributed table with shards on the coordinator
|
||||
--
|
||||
-- This test file has an alternative output because of the change in the
|
||||
-- display of SQL-standard function's arguments in INSERT/SELECT in PG15.
|
||||
-- The alternative output can be deleted when we drop support for PG14
|
||||
--
|
||||
CREATE SCHEMA cte_inline;
|
||||
SET search_path TO cte_inline;
|
||||
SET citus.next_shard_id TO 1960000;
|
||||
|
|
@ -857,10 +866,10 @@ DEBUG: CTE fist_table_cte is going to be inlined via distributed planning
|
|||
DEBUG: Router planner cannot handle multi-shard select queries
|
||||
DEBUG: performing repartitioned INSERT ... SELECT
|
||||
DEBUG: partitioning SELECT query by column index 0 with name 'key'
|
||||
DEBUG: distributed statement: INSERT INTO cte_inline.test_table_1960000 AS citus_table_alias (key, value) SELECT key, value FROM read_intermediate_results('{repartitioned_results_xxxxx_from_1960000_to_0}'::text[], 'binary'::citus_copy_format) intermediate_result(key integer, value text)
|
||||
DEBUG: distributed statement: INSERT INTO cte_inline.test_table_1960001 AS citus_table_alias (key, value) SELECT key, value FROM read_intermediate_results('{repartitioned_results_xxxxx_from_1960001_to_1}'::text[], 'binary'::citus_copy_format) intermediate_result(key integer, value text)
|
||||
DEBUG: distributed statement: INSERT INTO cte_inline.test_table_1960002 AS citus_table_alias (key, value) SELECT key, value FROM read_intermediate_results('{repartitioned_results_xxxxx_from_1960002_to_2}'::text[], 'binary'::citus_copy_format) intermediate_result(key integer, value text)
|
||||
DEBUG: distributed statement: INSERT INTO cte_inline.test_table_1960003 AS citus_table_alias (key, value) SELECT key, value FROM read_intermediate_results('{repartitioned_results_xxxxx_from_1960003_to_3}'::text[], 'binary'::citus_copy_format) intermediate_result(key integer, value text)
|
||||
DEBUG: distributed statement: INSERT INTO cte_inline.test_table_1960000 AS citus_table_alias (key, value) SELECT intermediate_result.key, intermediate_result.value FROM read_intermediate_results('{repartitioned_results_xxxxx_from_1960000_to_0}'::text[], 'binary'::citus_copy_format) intermediate_result(key integer, value text)
|
||||
DEBUG: distributed statement: INSERT INTO cte_inline.test_table_1960001 AS citus_table_alias (key, value) SELECT intermediate_result.key, intermediate_result.value FROM read_intermediate_results('{repartitioned_results_xxxxx_from_1960001_to_1}'::text[], 'binary'::citus_copy_format) intermediate_result(key integer, value text)
|
||||
DEBUG: distributed statement: INSERT INTO cte_inline.test_table_1960002 AS citus_table_alias (key, value) SELECT intermediate_result.key, intermediate_result.value FROM read_intermediate_results('{repartitioned_results_xxxxx_from_1960002_to_2}'::text[], 'binary'::citus_copy_format) intermediate_result(key integer, value text)
|
||||
DEBUG: distributed statement: INSERT INTO cte_inline.test_table_1960003 AS citus_table_alias (key, value) SELECT intermediate_result.key, intermediate_result.value FROM read_intermediate_results('{repartitioned_results_xxxxx_from_1960003_to_3}'::text[], 'binary'::citus_copy_format) intermediate_result(key integer, value text)
|
||||
-- the following INSERT..SELECT is even more interesting
|
||||
-- the CTE becomes pushdownable
|
||||
INSERT INTO test_table
|
||||
|
|
@ -871,10 +880,10 @@ WITH fist_table_cte AS
|
|||
FROM
|
||||
fist_table_cte;
|
||||
DEBUG: CTE fist_table_cte is going to be inlined via distributed planning
|
||||
DEBUG: distributed statement: INSERT INTO cte_inline.test_table_1960000 AS citus_table_alias (key, value) SELECT key, value FROM (SELECT test_table.key, test_table.value, test_table.other_value FROM cte_inline.test_table_1960000 test_table) fist_table_cte WHERE (key IS NOT NULL)
|
||||
DEBUG: distributed statement: INSERT INTO cte_inline.test_table_1960001 AS citus_table_alias (key, value) SELECT key, value FROM (SELECT test_table.key, test_table.value, test_table.other_value FROM cte_inline.test_table_1960001 test_table) fist_table_cte WHERE (key IS NOT NULL)
|
||||
DEBUG: distributed statement: INSERT INTO cte_inline.test_table_1960002 AS citus_table_alias (key, value) SELECT key, value FROM (SELECT test_table.key, test_table.value, test_table.other_value FROM cte_inline.test_table_1960002 test_table) fist_table_cte WHERE (key IS NOT NULL)
|
||||
DEBUG: distributed statement: INSERT INTO cte_inline.test_table_1960003 AS citus_table_alias (key, value) SELECT key, value FROM (SELECT test_table.key, test_table.value, test_table.other_value FROM cte_inline.test_table_1960003 test_table) fist_table_cte WHERE (key IS NOT NULL)
|
||||
DEBUG: distributed statement: INSERT INTO cte_inline.test_table_1960000 AS citus_table_alias (key, value) SELECT fist_table_cte.key, fist_table_cte.value FROM (SELECT test_table.key, test_table.value, test_table.other_value FROM cte_inline.test_table_1960000 test_table) fist_table_cte WHERE (fist_table_cte.key IS NOT NULL)
|
||||
DEBUG: distributed statement: INSERT INTO cte_inline.test_table_1960001 AS citus_table_alias (key, value) SELECT fist_table_cte.key, fist_table_cte.value FROM (SELECT test_table.key, test_table.value, test_table.other_value FROM cte_inline.test_table_1960001 test_table) fist_table_cte WHERE (fist_table_cte.key IS NOT NULL)
|
||||
DEBUG: distributed statement: INSERT INTO cte_inline.test_table_1960002 AS citus_table_alias (key, value) SELECT fist_table_cte.key, fist_table_cte.value FROM (SELECT test_table.key, test_table.value, test_table.other_value FROM cte_inline.test_table_1960002 test_table) fist_table_cte WHERE (fist_table_cte.key IS NOT NULL)
|
||||
DEBUG: distributed statement: INSERT INTO cte_inline.test_table_1960003 AS citus_table_alias (key, value) SELECT fist_table_cte.key, fist_table_cte.value FROM (SELECT test_table.key, test_table.value, test_table.other_value FROM cte_inline.test_table_1960003 test_table) fist_table_cte WHERE (fist_table_cte.key IS NOT NULL)
|
||||
-- update/delete/modifying ctes
|
||||
-- we don't support any cte inlining in modifications
|
||||
-- queries and modifying CTEs
|
||||
|
|
|
|||
File diff suppressed because it is too large
Load Diff
|
|
@ -1,3 +1,12 @@
|
|||
--
|
||||
-- INSERT_SELECT_REPARTITION
|
||||
--
|
||||
-- Test queries on a distributed table with shards on the coordinator
|
||||
--
|
||||
-- This test file has an alternative output because of the change in the
|
||||
-- display of SQL-standard function's arguments in INSERT/SELECT in PG15.
|
||||
-- The alternative output can be deleted when we drop support for PG14
|
||||
--
|
||||
-- tests behaviour of INSERT INTO ... SELECT with repartitioning
|
||||
CREATE SCHEMA insert_select_repartition;
|
||||
SET search_path TO 'insert_select_repartition';
|
||||
|
|
@ -29,10 +38,10 @@ HINT: Ensure the target table's partition column has a corresponding simple col
|
|||
DEBUG: Router planner cannot handle multi-shard select queries
|
||||
DEBUG: performing repartitioned INSERT ... SELECT
|
||||
DEBUG: partitioning SELECT query by column index 0 with name 'a'
|
||||
DEBUG: distributed statement: INSERT INTO insert_select_repartition.target_table_4213585 AS citus_table_alias (a) SELECT a FROM read_intermediate_results('{repartitioned_results_xxxxx_from_4213583_to_0,repartitioned_results_xxxxx_from_4213584_to_0}'::text[], 'binary'::citus_copy_format) intermediate_result(a integer)
|
||||
DEBUG: distributed statement: INSERT INTO insert_select_repartition.target_table_4213586 AS citus_table_alias (a) SELECT a FROM read_intermediate_results('{repartitioned_results_xxxxx_from_4213582_to_1}'::text[], 'binary'::citus_copy_format) intermediate_result(a integer)
|
||||
DEBUG: distributed statement: INSERT INTO insert_select_repartition.target_table_4213587 AS citus_table_alias (a) SELECT a FROM read_intermediate_results('{repartitioned_results_xxxxx_from_4213581_to_2,repartitioned_results_xxxxx_from_4213582_to_2,repartitioned_results_xxxxx_from_4213584_to_2}'::text[], 'binary'::citus_copy_format) intermediate_result(a integer)
|
||||
DEBUG: distributed statement: INSERT INTO insert_select_repartition.target_table_4213588 AS citus_table_alias (a) SELECT a FROM read_intermediate_results('{repartitioned_results_xxxxx_from_4213581_to_3}'::text[], 'binary'::citus_copy_format) intermediate_result(a integer)
|
||||
DEBUG: distributed statement: INSERT INTO insert_select_repartition.target_table_4213585 AS citus_table_alias (a) SELECT intermediate_result.a FROM read_intermediate_results('{repartitioned_results_xxxxx_from_4213583_to_0,repartitioned_results_xxxxx_from_4213584_to_0}'::text[], 'binary'::citus_copy_format) intermediate_result(a integer)
|
||||
DEBUG: distributed statement: INSERT INTO insert_select_repartition.target_table_4213586 AS citus_table_alias (a) SELECT intermediate_result.a FROM read_intermediate_results('{repartitioned_results_xxxxx_from_4213582_to_1}'::text[], 'binary'::citus_copy_format) intermediate_result(a integer)
|
||||
DEBUG: distributed statement: INSERT INTO insert_select_repartition.target_table_4213587 AS citus_table_alias (a) SELECT intermediate_result.a FROM read_intermediate_results('{repartitioned_results_xxxxx_from_4213581_to_2,repartitioned_results_xxxxx_from_4213582_to_2,repartitioned_results_xxxxx_from_4213584_to_2}'::text[], 'binary'::citus_copy_format) intermediate_result(a integer)
|
||||
DEBUG: distributed statement: INSERT INTO insert_select_repartition.target_table_4213588 AS citus_table_alias (a) SELECT intermediate_result.a FROM read_intermediate_results('{repartitioned_results_xxxxx_from_4213581_to_3}'::text[], 'binary'::citus_copy_format) intermediate_result(a integer)
|
||||
RESET client_min_messages;
|
||||
SELECT * FROM target_table WHERE a=-1 OR a=-3 OR a=-7 ORDER BY a;
|
||||
a
|
||||
|
|
@ -79,8 +88,8 @@ DETAIL: The target table's partition column should correspond to a partition co
|
|||
DEBUG: Router planner cannot handle multi-shard select queries
|
||||
DEBUG: performing repartitioned INSERT ... SELECT
|
||||
DEBUG: partitioning SELECT query by column index 2 with name 'key'
|
||||
DEBUG: distributed statement: INSERT INTO insert_select_repartition.target_table_4213591 AS citus_table_alias (f1, value, key) SELECT f1, value, key FROM read_intermediate_results('{repartitioned_results_xxxxx_from_4213589_to_0,repartitioned_results_xxxxx_from_4213590_to_0}'::text[], 'text'::citus_copy_format) intermediate_result(f1 integer, value integer, key insert_select_repartition.composite_key_type)
|
||||
DEBUG: distributed statement: INSERT INTO insert_select_repartition.target_table_4213592 AS citus_table_alias (f1, value, key) SELECT f1, value, key FROM read_intermediate_results('{repartitioned_results_xxxxx_from_4213589_to_1,repartitioned_results_xxxxx_from_4213590_to_1}'::text[], 'text'::citus_copy_format) intermediate_result(f1 integer, value integer, key insert_select_repartition.composite_key_type)
|
||||
DEBUG: distributed statement: INSERT INTO insert_select_repartition.target_table_4213591 AS citus_table_alias (f1, value, key) SELECT intermediate_result.f1, intermediate_result.value, intermediate_result.key FROM read_intermediate_results('{repartitioned_results_xxxxx_from_4213589_to_0,repartitioned_results_xxxxx_from_4213590_to_0}'::text[], 'text'::citus_copy_format) intermediate_result(f1 integer, value integer, key insert_select_repartition.composite_key_type)
|
||||
DEBUG: distributed statement: INSERT INTO insert_select_repartition.target_table_4213592 AS citus_table_alias (f1, value, key) SELECT intermediate_result.f1, intermediate_result.value, intermediate_result.key FROM read_intermediate_results('{repartitioned_results_xxxxx_from_4213589_to_1,repartitioned_results_xxxxx_from_4213590_to_1}'::text[], 'text'::citus_copy_format) intermediate_result(f1 integer, value integer, key insert_select_repartition.composite_key_type)
|
||||
RESET client_min_messages;
|
||||
SELECT * FROM target_table ORDER BY key;
|
||||
f1 | value | key
|
||||
|
|
@ -109,8 +118,8 @@ DETAIL: The target table's partition column should correspond to a partition co
|
|||
DEBUG: Router planner cannot handle multi-shard select queries
|
||||
DEBUG: performing repartitioned INSERT ... SELECT
|
||||
DEBUG: partitioning SELECT query by column index 2 with name 'key'
|
||||
DEBUG: distributed statement: INSERT INTO insert_select_repartition.target_table_4213591 AS citus_table_alias (f1, value, key) SELECT f1, value, key FROM read_intermediate_results('{repartitioned_results_xxxxx_from_4213589_to_0,repartitioned_results_xxxxx_from_4213590_to_0}'::text[], 'text'::citus_copy_format) intermediate_result(f1 integer, value integer, key insert_select_repartition.composite_key_type)
|
||||
DEBUG: distributed statement: INSERT INTO insert_select_repartition.target_table_4213592 AS citus_table_alias (f1, value, key) SELECT f1, value, key FROM read_intermediate_results('{repartitioned_results_xxxxx_from_4213589_to_1,repartitioned_results_xxxxx_from_4213590_to_1}'::text[], 'text'::citus_copy_format) intermediate_result(f1 integer, value integer, key insert_select_repartition.composite_key_type)
|
||||
DEBUG: distributed statement: INSERT INTO insert_select_repartition.target_table_4213591 AS citus_table_alias (f1, value, key) SELECT intermediate_result.f1, intermediate_result.value, intermediate_result.key FROM read_intermediate_results('{repartitioned_results_xxxxx_from_4213589_to_0,repartitioned_results_xxxxx_from_4213590_to_0}'::text[], 'text'::citus_copy_format) intermediate_result(f1 integer, value integer, key insert_select_repartition.composite_key_type)
|
||||
DEBUG: distributed statement: INSERT INTO insert_select_repartition.target_table_4213592 AS citus_table_alias (f1, value, key) SELECT intermediate_result.f1, intermediate_result.value, intermediate_result.key FROM read_intermediate_results('{repartitioned_results_xxxxx_from_4213589_to_1,repartitioned_results_xxxxx_from_4213590_to_1}'::text[], 'text'::citus_copy_format) intermediate_result(f1 integer, value integer, key insert_select_repartition.composite_key_type)
|
||||
RESET client_min_messages;
|
||||
SELECT * FROM target_table ORDER BY key;
|
||||
f1 | value | key
|
||||
|
|
@ -133,8 +142,8 @@ DETAIL: The target table's partition column should correspond to a partition co
|
|||
DEBUG: Router planner cannot handle multi-shard select queries
|
||||
DEBUG: performing repartitioned INSERT ... SELECT
|
||||
DEBUG: partitioning SELECT query by column index 1 with name 'key'
|
||||
DEBUG: distributed statement: INSERT INTO insert_select_repartition.target_table_4213591 AS citus_table_alias (f1, key) SELECT f1, key FROM read_intermediate_results('{repartitioned_results_xxxxx_from_4213589_to_0,repartitioned_results_xxxxx_from_4213590_to_0}'::text[], 'text'::citus_copy_format) intermediate_result(f1 integer, key insert_select_repartition.composite_key_type)
|
||||
DEBUG: distributed statement: INSERT INTO insert_select_repartition.target_table_4213592 AS citus_table_alias (f1, key) SELECT f1, key FROM read_intermediate_results('{repartitioned_results_xxxxx_from_4213589_to_1,repartitioned_results_xxxxx_from_4213590_to_1}'::text[], 'text'::citus_copy_format) intermediate_result(f1 integer, key insert_select_repartition.composite_key_type)
|
||||
DEBUG: distributed statement: INSERT INTO insert_select_repartition.target_table_4213591 AS citus_table_alias (f1, key) SELECT intermediate_result.f1, intermediate_result.key FROM read_intermediate_results('{repartitioned_results_xxxxx_from_4213589_to_0,repartitioned_results_xxxxx_from_4213590_to_0}'::text[], 'text'::citus_copy_format) intermediate_result(f1 integer, key insert_select_repartition.composite_key_type)
|
||||
DEBUG: distributed statement: INSERT INTO insert_select_repartition.target_table_4213592 AS citus_table_alias (f1, key) SELECT intermediate_result.f1, intermediate_result.key FROM read_intermediate_results('{repartitioned_results_xxxxx_from_4213589_to_1,repartitioned_results_xxxxx_from_4213590_to_1}'::text[], 'text'::citus_copy_format) intermediate_result(f1 integer, key insert_select_repartition.composite_key_type)
|
||||
RESET client_min_messages;
|
||||
SELECT * FROM target_table ORDER BY key;
|
||||
f1 | value | key
|
||||
|
|
@ -159,8 +168,8 @@ DETAIL: The target table's partition column should correspond to a partition co
|
|||
DEBUG: Router planner cannot handle multi-shard select queries
|
||||
DEBUG: performing repartitioned INSERT ... SELECT
|
||||
DEBUG: partitioning SELECT query by column index 1 with name 'key'
|
||||
DEBUG: distributed statement: INSERT INTO insert_select_repartition.target_table_4213591 AS citus_table_alias (f1, key) SELECT f1, key FROM read_intermediate_results('{repartitioned_results_xxxxx_from_4213589_to_0}'::text[], 'text'::citus_copy_format) intermediate_result(f1 integer, key insert_select_repartition.composite_key_type) ON CONFLICT(key) DO UPDATE SET f1 = 1
|
||||
DEBUG: distributed statement: INSERT INTO insert_select_repartition.target_table_4213592 AS citus_table_alias (f1, key) SELECT f1, key FROM read_intermediate_results('{repartitioned_results_xxxxx_from_4213589_to_1,repartitioned_results_xxxxx_from_4213590_to_1}'::text[], 'text'::citus_copy_format) intermediate_result(f1 integer, key insert_select_repartition.composite_key_type) ON CONFLICT(key) DO UPDATE SET f1 = 1
|
||||
DEBUG: distributed statement: INSERT INTO insert_select_repartition.target_table_4213591 AS citus_table_alias (f1, key) SELECT intermediate_result.f1, intermediate_result.key FROM read_intermediate_results('{repartitioned_results_xxxxx_from_4213589_to_0}'::text[], 'text'::citus_copy_format) intermediate_result(f1 integer, key insert_select_repartition.composite_key_type) ON CONFLICT(key) DO UPDATE SET f1 = 1
|
||||
DEBUG: distributed statement: INSERT INTO insert_select_repartition.target_table_4213592 AS citus_table_alias (f1, key) SELECT intermediate_result.f1, intermediate_result.key FROM read_intermediate_results('{repartitioned_results_xxxxx_from_4213589_to_1,repartitioned_results_xxxxx_from_4213590_to_1}'::text[], 'text'::citus_copy_format) intermediate_result(f1 integer, key insert_select_repartition.composite_key_type) ON CONFLICT(key) DO UPDATE SET f1 = 1
|
||||
RESET client_min_messages;
|
||||
SELECT * FROM target_table ORDER BY key;
|
||||
f1 | value | key
|
||||
|
|
@ -209,8 +218,8 @@ DETAIL: The data type of the target table's partition column should exactly mat
|
|||
DEBUG: Router planner cannot handle multi-shard select queries
|
||||
DEBUG: performing repartitioned INSERT ... SELECT
|
||||
DEBUG: partitioning SELECT query by column index 0 with name 'col_1'
|
||||
DEBUG: distributed statement: INSERT INTO insert_select_repartition.target_table_4213593 AS citus_table_alias (col_1, col_2) SELECT col_1, col_2 FROM read_intermediate_results('{repartitioned_results_xxxxx_from_4213597_to_0,repartitioned_results_xxxxx_from_4213600_to_0}'::text[], 'binary'::citus_copy_format) intermediate_result(col_1 integer, col_2 integer) ON CONFLICT(col_1) DO UPDATE SET col_2 = excluded.col_2
|
||||
DEBUG: distributed statement: INSERT INTO insert_select_repartition.target_table_4213594 AS citus_table_alias (col_1, col_2) SELECT col_1, col_2 FROM read_intermediate_results('{repartitioned_results_xxxxx_from_4213599_to_1}'::text[], 'binary'::citus_copy_format) intermediate_result(col_1 integer, col_2 integer) ON CONFLICT(col_1) DO UPDATE SET col_2 = excluded.col_2
|
||||
DEBUG: distributed statement: INSERT INTO insert_select_repartition.target_table_4213593 AS citus_table_alias (col_1, col_2) SELECT intermediate_result.col_1, intermediate_result.col_2 FROM read_intermediate_results('{repartitioned_results_xxxxx_from_4213597_to_0,repartitioned_results_xxxxx_from_4213600_to_0}'::text[], 'binary'::citus_copy_format) intermediate_result(col_1 integer, col_2 integer) ON CONFLICT(col_1) DO UPDATE SET col_2 = excluded.col_2
|
||||
DEBUG: distributed statement: INSERT INTO insert_select_repartition.target_table_4213594 AS citus_table_alias (col_1, col_2) SELECT intermediate_result.col_1, intermediate_result.col_2 FROM read_intermediate_results('{repartitioned_results_xxxxx_from_4213599_to_1}'::text[], 'binary'::citus_copy_format) intermediate_result(col_1 integer, col_2 integer) ON CONFLICT(col_1) DO UPDATE SET col_2 = excluded.col_2
|
||||
RESET client_min_messages;
|
||||
SELECT * FROM target_table ORDER BY 1;
|
||||
col_1 | col_2
|
||||
|
|
@ -474,7 +483,7 @@ WITH c AS (
|
|||
SELECT mapped_key, c FROM source_table
|
||||
RETURNING *)
|
||||
SELECT * FROM c ORDER by a;
|
||||
DEBUG: generating subplan XXX_1 for CTE c: INSERT INTO insert_select_repartition.target_table (a, b) SELECT mapped_key, c FROM insert_select_repartition.source_table RETURNING target_table.a, target_table.b
|
||||
DEBUG: generating subplan XXX_1 for CTE c: INSERT INTO insert_select_repartition.target_table (a, b) SELECT source_table.mapped_key, source_table.c FROM insert_select_repartition.source_table RETURNING target_table.a, target_table.b
|
||||
DEBUG: cannot perform distributed INSERT INTO ... SELECT because the partition columns in the source table and subquery do not match
|
||||
DEBUG: Plan XXX query after replacing subqueries and CTEs: SELECT a, b FROM (SELECT intermediate_result.a, intermediate_result.b FROM read_intermediate_result('XXX_1'::text, 'binary'::citus_copy_format) intermediate_result(a integer, b integer[])) c ORDER BY a
|
||||
DEBUG: performing repartitioned INSERT ... SELECT
|
||||
|
|
@ -583,9 +592,9 @@ DEBUG: cannot perform distributed INSERT INTO ... SELECT because the partition
|
|||
DEBUG: Router planner cannot handle multi-shard select queries
|
||||
DEBUG: performing repartitioned INSERT ... SELECT
|
||||
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,repartitioned_results_xxxxx_from_4213609_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_4213606_to_2,repartitioned_results_xxxxx_from_4213607_to_2}'::text[], 'binary'::citus_copy_format) intermediate_result(a integer, b integer)
|
||||
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,repartitioned_results_xxxxx_from_4213609_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_4213606_to_2,repartitioned_results_xxxxx_from_4213607_to_2}'::text[], 'binary'::citus_copy_format) intermediate_result(a integer, b integer)
|
||||
RESET client_min_messages;
|
||||
SELECT * FROM target_table ORDER BY a;
|
||||
a | b
|
||||
|
|
@ -748,7 +757,7 @@ WITH r AS (
|
|||
INSERT INTO target_table SELECT source_table.a, max(source_table.b) FROM source_table NATURAL JOIN r GROUP BY source_table.a;
|
||||
DEBUG: INSERT target table and the source relation of the SELECT partition column value must be colocated in distributed INSERT ... SELECT
|
||||
DEBUG: only SELECT, UPDATE, or DELETE common table expressions may be router planned
|
||||
DEBUG: generating subplan XXX_1 for CTE r: INSERT INTO insert_select_repartition.target_table (a, b) SELECT a, b FROM insert_select_repartition.source_table RETURNING target_table.a, target_table.b
|
||||
DEBUG: generating subplan XXX_1 for CTE r: INSERT INTO insert_select_repartition.target_table (a, b) SELECT source_table.a, source_table.b FROM insert_select_repartition.source_table RETURNING target_table.a, target_table.b
|
||||
DEBUG: INSERT target table and the source relation of the SELECT partition column value must be colocated in distributed INSERT ... SELECT
|
||||
DEBUG: Router planner cannot handle multi-shard select queries
|
||||
DEBUG: Plan XXX query after replacing subqueries and CTEs: SELECT a, max AS b FROM (SELECT source_table.a, max(source_table.b) AS max FROM (insert_select_repartition.source_table JOIN (SELECT intermediate_result.a, intermediate_result.b FROM read_intermediate_result('XXX_1'::text, 'binary'::citus_copy_format) intermediate_result(a integer, b integer)) r USING (a, b)) GROUP BY source_table.a) citus_insert_select_subquery
|
||||
|
|
@ -756,13 +765,13 @@ DEBUG: Router planner cannot handle multi-shard select queries
|
|||
DEBUG: performing repartitioned INSERT ... SELECT
|
||||
DEBUG: performing repartitioned INSERT ... SELECT
|
||||
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) RETURNING citus_table_alias.a, citus_table_alias.b
|
||||
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) RETURNING citus_table_alias.a, citus_table_alias.b
|
||||
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) RETURNING citus_table_alias.a, citus_table_alias.b
|
||||
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) RETURNING citus_table_alias.a, citus_table_alias.b
|
||||
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) RETURNING citus_table_alias.a, citus_table_alias.b
|
||||
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) RETURNING citus_table_alias.a, citus_table_alias.b
|
||||
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)
|
||||
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 client_min_messages;
|
||||
SELECT * FROM target_table ORDER BY a, b;
|
||||
a | b
|
||||
|
|
@ -892,7 +901,7 @@ DEBUG: cannot perform distributed INSERT INTO ... SELECT because the partition
|
|||
DEBUG: Router planner cannot handle multi-shard select queries
|
||||
DEBUG: performing repartitioned INSERT ... SELECT
|
||||
DEBUG: partitioning SELECT query by column index 0 with name 'a'
|
||||
DEBUG: distributed statement: INSERT INTO insert_select_repartition.target_table_4213617 AS citus_table_alias (a, b) SELECT a, b FROM read_intermediate_results('{repartitioned_results_xxxxx_from_4213613_to_0,repartitioned_results_xxxxx_from_4213614_to_0,repartitioned_results_xxxxx_from_4213615_to_0,repartitioned_results_xxxxx_from_4213616_to_0}'::text[], 'binary'::citus_copy_format) intermediate_result(a integer, b integer)
|
||||
DEBUG: distributed statement: INSERT INTO insert_select_repartition.target_table_4213617 AS citus_table_alias (a, b) SELECT intermediate_result.a, intermediate_result.b FROM read_intermediate_results('{repartitioned_results_xxxxx_from_4213613_to_0,repartitioned_results_xxxxx_from_4213614_to_0,repartitioned_results_xxxxx_from_4213615_to_0,repartitioned_results_xxxxx_from_4213616_to_0}'::text[], 'binary'::citus_copy_format) intermediate_result(a integer, b integer)
|
||||
RESET client_min_messages;
|
||||
SELECT * FROM target_table ORDER BY a, b;
|
||||
a | b
|
||||
|
|
@ -1003,9 +1012,9 @@ DEBUG: INSERT target table and the source relation of the SELECT partition colu
|
|||
DEBUG: Router planner cannot handle multi-shard select queries
|
||||
DEBUG: performing repartitioned INSERT ... SELECT
|
||||
DEBUG: partitioning SELECT query by column index 0 with name 'a'
|
||||
DEBUG: distributed statement: INSERT INTO insert_select_repartition.target_table_4213633 AS citus_table_alias (a, b, c, d) SELECT a, b, c, d FROM read_intermediate_results('{repartitioned_results_xxxxx_from_4213629_to_0,repartitioned_results_xxxxx_from_4213630_to_0}'::text[], 'binary'::citus_copy_format) intermediate_result(a integer, b integer, c integer, d integer)
|
||||
DEBUG: distributed statement: INSERT INTO insert_select_repartition.target_table_4213634 AS citus_table_alias (a, b, c, d) SELECT a, b, c, d FROM read_intermediate_results('{repartitioned_results_xxxxx_from_4213630_to_1,repartitioned_results_xxxxx_from_4213631_to_1}'::text[], 'binary'::citus_copy_format) intermediate_result(a integer, b integer, c integer, d integer)
|
||||
DEBUG: distributed statement: INSERT INTO insert_select_repartition.target_table_4213635 AS citus_table_alias (a, b, c, d) SELECT a, b, c, d FROM read_intermediate_results('{repartitioned_results_xxxxx_from_4213632_to_2}'::text[], 'binary'::citus_copy_format) intermediate_result(a integer, b integer, c integer, d integer)
|
||||
DEBUG: distributed statement: INSERT INTO insert_select_repartition.target_table_4213633 AS citus_table_alias (a, b, c, d) SELECT intermediate_result.a, intermediate_result.b, intermediate_result.c, intermediate_result.d FROM read_intermediate_results('{repartitioned_results_xxxxx_from_4213629_to_0,repartitioned_results_xxxxx_from_4213630_to_0}'::text[], 'binary'::citus_copy_format) intermediate_result(a integer, b integer, c integer, d integer)
|
||||
DEBUG: distributed statement: INSERT INTO insert_select_repartition.target_table_4213634 AS citus_table_alias (a, b, c, d) SELECT intermediate_result.a, intermediate_result.b, intermediate_result.c, intermediate_result.d FROM read_intermediate_results('{repartitioned_results_xxxxx_from_4213630_to_1,repartitioned_results_xxxxx_from_4213631_to_1}'::text[], 'binary'::citus_copy_format) intermediate_result(a integer, b integer, c integer, d integer)
|
||||
DEBUG: distributed statement: INSERT INTO insert_select_repartition.target_table_4213635 AS citus_table_alias (a, b, c, d) SELECT intermediate_result.a, intermediate_result.b, intermediate_result.c, intermediate_result.d FROM read_intermediate_results('{repartitioned_results_xxxxx_from_4213632_to_2}'::text[], 'binary'::citus_copy_format) intermediate_result(a integer, b integer, c integer, d integer)
|
||||
RESET client_min_messages;
|
||||
SELECT count(*) FROM target_table;
|
||||
count
|
||||
|
|
@ -1207,8 +1216,8 @@ DEBUG: INSERT target table and the source relation of the SELECT partition colu
|
|||
DEBUG: Router planner cannot handle multi-shard select queries
|
||||
DEBUG: performing repartitioned INSERT ... SELECT
|
||||
DEBUG: partitioning SELECT query by column index 0 with name 'c1'
|
||||
DEBUG: distributed statement: INSERT INTO insert_select_repartition.target_table_4213639 AS enriched (c1, c2, c3, c4, c5, c6, cardinality, sum) SELECT c1, c2, c3, c4, c5, c6, cardinality, sum FROM read_intermediate_results('{repartitioned_results_xxxxx_from_4213644_to_0}'::text[], 'binary'::citus_copy_format) intermediate_result(c1 integer, c2 integer, c3 timestamp without time zone, c4 integer, c5 integer, c6 integer[], cardinality integer, sum integer) ON CONFLICT(c1, c2, c3, c4, c5, c6) DO UPDATE SET cardinality = (enriched.cardinality OPERATOR(pg_catalog.+) excluded.cardinality), sum = (enriched.sum OPERATOR(pg_catalog.+) excluded.sum)
|
||||
DEBUG: distributed statement: INSERT INTO insert_select_repartition.target_table_4213641 AS enriched (c1, c2, c3, c4, c5, c6, cardinality, sum) SELECT c1, c2, c3, c4, c5, c6, cardinality, sum FROM read_intermediate_results('{repartitioned_results_xxxxx_from_4213645_to_2}'::text[], 'binary'::citus_copy_format) intermediate_result(c1 integer, c2 integer, c3 timestamp without time zone, c4 integer, c5 integer, c6 integer[], cardinality integer, sum integer) ON CONFLICT(c1, c2, c3, c4, c5, c6) DO UPDATE SET cardinality = (enriched.cardinality OPERATOR(pg_catalog.+) excluded.cardinality), sum = (enriched.sum OPERATOR(pg_catalog.+) excluded.sum)
|
||||
DEBUG: distributed statement: INSERT INTO insert_select_repartition.target_table_4213639 AS enriched (c1, c2, c3, c4, c5, c6, cardinality, sum) SELECT intermediate_result.c1, intermediate_result.c2, intermediate_result.c3, intermediate_result.c4, intermediate_result.c5, intermediate_result.c6, intermediate_result.cardinality, intermediate_result.sum FROM read_intermediate_results('{repartitioned_results_xxxxx_from_4213644_to_0}'::text[], 'binary'::citus_copy_format) intermediate_result(c1 integer, c2 integer, c3 timestamp without time zone, c4 integer, c5 integer, c6 integer[], cardinality integer, sum integer) ON CONFLICT(c1, c2, c3, c4, c5, c6) DO UPDATE SET cardinality = (enriched.cardinality OPERATOR(pg_catalog.+) excluded.cardinality), sum = (enriched.sum OPERATOR(pg_catalog.+) excluded.sum)
|
||||
DEBUG: distributed statement: INSERT INTO insert_select_repartition.target_table_4213641 AS enriched (c1, c2, c3, c4, c5, c6, cardinality, sum) SELECT intermediate_result.c1, intermediate_result.c2, intermediate_result.c3, intermediate_result.c4, intermediate_result.c5, intermediate_result.c6, intermediate_result.cardinality, intermediate_result.sum FROM read_intermediate_results('{repartitioned_results_xxxxx_from_4213645_to_2}'::text[], 'binary'::citus_copy_format) intermediate_result(c1 integer, c2 integer, c3 timestamp without time zone, c4 integer, c5 integer, c6 integer[], cardinality integer, sum integer) ON CONFLICT(c1, c2, c3, c4, c5, c6) DO UPDATE SET cardinality = (enriched.cardinality OPERATOR(pg_catalog.+) excluded.cardinality), sum = (enriched.sum OPERATOR(pg_catalog.+) excluded.sum)
|
||||
RESET client_min_messages;
|
||||
EXPLAIN (COSTS OFF) INSERT INTO target_table AS enriched(c1, c2, c3, c4, c5, c6, cardinality, sum)
|
||||
SELECT c1, c2, c3, c4, -1::float AS c5,
|
||||
|
|
|
|||
File diff suppressed because it is too large
Load Diff
|
|
@ -1,3 +1,12 @@
|
|||
--
|
||||
-- INTERMEDIATE_RESULT_PRUNING
|
||||
--
|
||||
-- Test queries on a distributed table with shards on the coordinator
|
||||
--
|
||||
-- This test file has an alternative output because of the change in the
|
||||
-- display of SQL-standard function's arguments in INSERT/SELECT in PG15.
|
||||
-- The alternative output can be deleted when we drop support for PG14
|
||||
--
|
||||
CREATE SCHEMA intermediate_result_pruning;
|
||||
SET search_path TO intermediate_result_pruning;
|
||||
SET citus.log_intermediate_results TO TRUE;
|
||||
|
|
@ -1039,7 +1048,7 @@ inserts AS MATERIALIZED (
|
|||
RETURNING *
|
||||
) SELECT count(*) FROM inserts;
|
||||
DEBUG: generating subplan XXX_1 for CTE stats: SELECT count(key) AS m FROM intermediate_result_pruning.table_3
|
||||
DEBUG: generating subplan XXX_2 for CTE inserts: INSERT INTO intermediate_result_pruning.table_2 (key, value) SELECT key, count(*) AS count FROM intermediate_result_pruning.table_1 WHERE (key OPERATOR(pg_catalog.>) (SELECT stats.m FROM (SELECT intermediate_result.m FROM read_intermediate_result('XXX_1'::text, 'binary'::citus_copy_format) intermediate_result(m bigint)) stats)) GROUP BY key HAVING (count(*) OPERATOR(pg_catalog.<) (SELECT stats.m FROM (SELECT intermediate_result.m FROM read_intermediate_result('XXX_1'::text, 'binary'::citus_copy_format) intermediate_result(m bigint)) stats)) LIMIT 1 RETURNING table_2.key, table_2.value
|
||||
DEBUG: generating subplan XXX_2 for CTE inserts: INSERT INTO intermediate_result_pruning.table_2 (key, value) SELECT table_1.key, count(*) AS count FROM intermediate_result_pruning.table_1 WHERE (table_1.key OPERATOR(pg_catalog.>) (SELECT stats.m FROM (SELECT intermediate_result.m FROM read_intermediate_result('XXX_1'::text, 'binary'::citus_copy_format) intermediate_result(m bigint)) stats)) GROUP BY table_1.key HAVING (count(*) OPERATOR(pg_catalog.<) (SELECT stats.m FROM (SELECT intermediate_result.m FROM read_intermediate_result('XXX_1'::text, 'binary'::citus_copy_format) intermediate_result(m bigint)) stats)) LIMIT 1 RETURNING table_2.key, table_2.value
|
||||
DEBUG: LIMIT clauses are not allowed in distributed INSERT ... SELECT queries
|
||||
DEBUG: push down of limit count: 1
|
||||
DEBUG: Plan XXX query after replacing subqueries and CTEs: SELECT count(*) AS count FROM (SELECT intermediate_result.key, intermediate_result.value FROM read_intermediate_result('XXX_2'::text, 'binary'::citus_copy_format) intermediate_result(key integer, value text)) inserts
|
||||
|
|
|
|||
File diff suppressed because it is too large
Load Diff
|
|
@ -1,3 +1,12 @@
|
|||
--
|
||||
-- LOCAL_SHARD_EXECUTION
|
||||
--
|
||||
-- Test queries on a distributed table with shards on the coordinator
|
||||
--
|
||||
-- This test file has an alternative output because of the change in the
|
||||
-- display of SQL-standard function's arguments in INSERT/SELECT in PG15.
|
||||
-- The alternative output can be deleted when we drop support for PG14
|
||||
--
|
||||
CREATE SCHEMA local_shard_execution;
|
||||
SET search_path TO local_shard_execution;
|
||||
SET citus.shard_count TO 4;
|
||||
|
|
@ -288,7 +297,7 @@ RETURNING *;
|
|||
INSERT INTO distributed_table SELECT * FROM distributed_table WHERE key = 1 OFFSET 0 ON CONFLICT DO NOTHING;
|
||||
NOTICE: executing the command locally: SELECT key, value, age FROM local_shard_execution.distributed_table_1470001 distributed_table WHERE (key OPERATOR(pg_catalog.=) 1) OFFSET 0
|
||||
NOTICE: executing the copy locally for colocated file with shard xxxxx
|
||||
NOTICE: executing the command locally: INSERT INTO local_shard_execution.distributed_table_1470001 AS citus_table_alias (key, value, age) SELECT key, value, age FROM read_intermediate_result('insert_select_XXX_1470001'::text, 'binary'::citus_copy_format) intermediate_result(key integer, value text, age bigint) ON CONFLICT DO NOTHING
|
||||
NOTICE: executing the command locally: INSERT INTO local_shard_execution.distributed_table_1470001 AS citus_table_alias (key, value, age) SELECT intermediate_result.key, intermediate_result.value, intermediate_result.age FROM read_intermediate_result('insert_select_XXX_1470001'::text, 'binary'::citus_copy_format) intermediate_result(key integer, value text, age bigint) ON CONFLICT DO NOTHING
|
||||
INSERT INTO distributed_table SELECT 1, '1',15 FROM distributed_table WHERE key = 2 LIMIT 1 ON CONFLICT DO NOTHING;
|
||||
-- sanity check: multi-shard INSERT..SELECT pushdown goes through distributed execution
|
||||
INSERT INTO distributed_table SELECT * FROM distributed_table ON CONFLICT DO NOTHING;
|
||||
|
|
@ -800,7 +809,7 @@ NOTICE: executing the copy locally for shard xxxxx
|
|||
INSERT INTO distributed_table (key) SELECT -key FROM distributed_table;
|
||||
NOTICE: executing the command locally: SELECT partition_index, 'repartitioned_results_xxxxx_from_1470001_to' || '_' || partition_index::text , rows_written FROM worker_partition_query_result('repartitioned_results_xxxxx_from_1470001_to','SELECT (OPERATOR(pg_catalog.-) key) AS key FROM local_shard_execution.distributed_table_1470001 distributed_table WHERE true',0,'hash','{-2147483648,-1073741824,0,1073741824}'::text[],'{-1073741825,-1,1073741823,2147483647}'::text[],true) WHERE rows_written > 0
|
||||
NOTICE: executing the command locally: SELECT partition_index, 'repartitioned_results_xxxxx_from_1470003_to' || '_' || partition_index::text , rows_written FROM worker_partition_query_result('repartitioned_results_xxxxx_from_1470003_to','SELECT (OPERATOR(pg_catalog.-) key) AS key FROM local_shard_execution.distributed_table_1470003 distributed_table WHERE true',0,'hash','{-2147483648,-1073741824,0,1073741824}'::text[],'{-1073741825,-1,1073741823,2147483647}'::text[],true) WHERE rows_written > 0
|
||||
NOTICE: executing the command locally: INSERT INTO local_shard_execution.distributed_table_1470001 AS citus_table_alias (key) SELECT key FROM read_intermediate_results('{repartitioned_results_xxxxx_from_1470003_to_0}'::text[], 'binary'::citus_copy_format) intermediate_result(key integer)
|
||||
NOTICE: executing the command locally: INSERT INTO local_shard_execution.distributed_table_1470001 AS citus_table_alias (key) SELECT intermediate_result.key FROM read_intermediate_results('{repartitioned_results_xxxxx_from_1470003_to_0}'::text[], 'binary'::citus_copy_format) intermediate_result(key integer)
|
||||
SELECT count(*) FROM distributed_table WHERE key = -6;
|
||||
NOTICE: executing the command locally: SELECT count(*) AS count FROM local_shard_execution.distributed_table_1470001 distributed_table WHERE (key OPERATOR(pg_catalog.=) '-6'::integer)
|
||||
count
|
||||
|
|
|
|||
File diff suppressed because it is too large
Load Diff
|
|
@ -1,3 +1,12 @@
|
|||
--
|
||||
-- LOCAL_SHARD_EXECUTION_REPLICATED
|
||||
--
|
||||
-- Test queries on a distributed table with shards on the coordinator
|
||||
--
|
||||
-- This test file has an alternative output because of the change in the
|
||||
-- display of SQL-standard function's arguments in INSERT/SELECT in PG15.
|
||||
-- The alternative output can be deleted when we drop support for PG14
|
||||
--
|
||||
CREATE SCHEMA local_shard_execution_replicated;
|
||||
SET search_path TO local_shard_execution_replicated;
|
||||
SET citus.shard_count TO 4;
|
||||
|
|
@ -225,7 +234,7 @@ RETURNING *;
|
|||
INSERT INTO distributed_table SELECT * FROM distributed_table WHERE key = 1 OFFSET 0 ON CONFLICT DO NOTHING;
|
||||
NOTICE: executing the command locally: SELECT key, value, age FROM local_shard_execution_replicated.distributed_table_1500001 distributed_table WHERE (key OPERATOR(pg_catalog.=) 1) OFFSET 0
|
||||
NOTICE: executing the copy locally for colocated file with shard xxxxx
|
||||
NOTICE: executing the command locally: INSERT INTO local_shard_execution_replicated.distributed_table_1500001 AS citus_table_alias (key, value, age) SELECT key, value, age FROM read_intermediate_result('insert_select_XXX_1500001'::text, 'binary'::citus_copy_format) intermediate_result(key integer, value text, age bigint) ON CONFLICT DO NOTHING
|
||||
NOTICE: executing the command locally: INSERT INTO local_shard_execution_replicated.distributed_table_1500001 AS citus_table_alias (key, value, age) SELECT intermediate_result.key, intermediate_result.value, intermediate_result.age FROM read_intermediate_result('insert_select_XXX_1500001'::text, 'binary'::citus_copy_format) intermediate_result(key integer, value text, age bigint) ON CONFLICT DO NOTHING
|
||||
INSERT INTO distributed_table SELECT 1, '1',15 FROM distributed_table WHERE key = 2 LIMIT 1 ON CONFLICT DO NOTHING;
|
||||
NOTICE: executing the command locally: SELECT 1 AS key, '1'::text AS value, int8(15) AS age FROM local_shard_execution_replicated.distributed_table_1500004 distributed_table WHERE (key OPERATOR(pg_catalog.=) 2) LIMIT 1
|
||||
-- sanity check: multi-shard INSERT..SELECT pushdown goes through distributed execution
|
||||
|
|
@ -764,8 +773,8 @@ NOTICE: executing the command locally: SELECT partition_index, 'repartitioned_r
|
|||
NOTICE: executing the command locally: SELECT partition_index, 'repartitioned_results_xxxxx_from_1500002_to' || '_' || partition_index::text , rows_written FROM worker_partition_query_result('repartitioned_results_xxxxx_from_1500002_to','SELECT (OPERATOR(pg_catalog.-) key) AS key FROM local_shard_execution_replicated.distributed_table_1500002 distributed_table WHERE true',0,'hash','{-2147483648,-1073741824,0,1073741824}'::text[],'{-1073741825,-1,1073741823,2147483647}'::text[],true) WHERE rows_written > 0
|
||||
NOTICE: executing the command locally: SELECT partition_index, 'repartitioned_results_xxxxx_from_1500003_to' || '_' || partition_index::text , rows_written FROM worker_partition_query_result('repartitioned_results_xxxxx_from_1500003_to','SELECT (OPERATOR(pg_catalog.-) key) AS key FROM local_shard_execution_replicated.distributed_table_1500003 distributed_table WHERE true',0,'hash','{-2147483648,-1073741824,0,1073741824}'::text[],'{-1073741825,-1,1073741823,2147483647}'::text[],true) WHERE rows_written > 0
|
||||
NOTICE: executing the command locally: SELECT partition_index, 'repartitioned_results_xxxxx_from_1500004_to' || '_' || partition_index::text , rows_written FROM worker_partition_query_result('repartitioned_results_xxxxx_from_1500004_to','SELECT (OPERATOR(pg_catalog.-) key) AS key FROM local_shard_execution_replicated.distributed_table_1500004 distributed_table WHERE true',0,'hash','{-2147483648,-1073741824,0,1073741824}'::text[],'{-1073741825,-1,1073741823,2147483647}'::text[],true) WHERE rows_written > 0
|
||||
NOTICE: executing the command locally: INSERT INTO local_shard_execution_replicated.distributed_table_1500001 AS citus_table_alias (key) SELECT key FROM read_intermediate_results('{repartitioned_results_xxxxx_from_1500003_to_0}'::text[], 'binary'::citus_copy_format) intermediate_result(key integer)
|
||||
NOTICE: executing the command locally: INSERT INTO local_shard_execution_replicated.distributed_table_1500004 AS citus_table_alias (key) SELECT key FROM read_intermediate_results('{repartitioned_results_xxxxx_from_1500004_to_3}'::text[], 'binary'::citus_copy_format) intermediate_result(key integer)
|
||||
NOTICE: executing the command locally: INSERT INTO local_shard_execution_replicated.distributed_table_1500001 AS citus_table_alias (key) SELECT intermediate_result.key FROM read_intermediate_results('{repartitioned_results_xxxxx_from_1500003_to_0}'::text[], 'binary'::citus_copy_format) intermediate_result(key integer)
|
||||
NOTICE: executing the command locally: INSERT INTO local_shard_execution_replicated.distributed_table_1500004 AS citus_table_alias (key) SELECT intermediate_result.key FROM read_intermediate_results('{repartitioned_results_xxxxx_from_1500004_to_3}'::text[], 'binary'::citus_copy_format) intermediate_result(key integer)
|
||||
SELECT count(*) FROM distributed_table WHERE key = -6;
|
||||
NOTICE: executing the command locally: SELECT count(*) AS count FROM local_shard_execution_replicated.distributed_table_1500001 distributed_table WHERE (key OPERATOR(pg_catalog.=) '-6'::integer)
|
||||
count
|
||||
|
|
|
|||
File diff suppressed because it is too large
Load Diff
|
|
@ -1,3 +1,12 @@
|
|||
--
|
||||
-- MULTI_MX_INSERT_SELECT_REPARTITION
|
||||
--
|
||||
-- Test queries on a distributed table with shards on the coordinator
|
||||
--
|
||||
-- This test file has an alternative output because of the change in the
|
||||
-- display of SQL-standard function's arguments in INSERT/SELECT in PG15.
|
||||
-- The alternative output can be deleted when we drop support for PG14
|
||||
--
|
||||
-- Test behaviour of repartitioned INSERT ... SELECT in MX setup
|
||||
CREATE SCHEMA multi_mx_insert_select_repartition;
|
||||
SET search_path TO multi_mx_insert_select_repartition;
|
||||
|
|
@ -92,8 +101,8 @@ NOTICE: executing the command locally: SELECT count(*) AS count FROM multi_mx_i
|
|||
NOTICE: executing the command locally: SELECT partition_index, 'repartitioned_results_xxxxx_from_4213581_to' || '_' || partition_index::text , rows_written FROM worker_partition_query_result('repartitioned_results_xxxxx_from_4213581_to','SELECT (a OPERATOR(pg_catalog.*) 2) AS a FROM multi_mx_insert_select_repartition.source_table_4213581 source_table WHERE true',0,'hash','{-2147483648,-715827883,715827882}'::text[],'{-715827884,715827881,2147483647}'::text[],true) WHERE rows_written > 0
|
||||
NOTICE: executing the command locally: SELECT partition_index, 'repartitioned_results_xxxxx_from_4213583_to' || '_' || partition_index::text , rows_written FROM worker_partition_query_result('repartitioned_results_xxxxx_from_4213583_to','SELECT (a OPERATOR(pg_catalog.*) 2) AS a FROM multi_mx_insert_select_repartition.source_table_4213583 source_table WHERE true',0,'hash','{-2147483648,-715827883,715827882}'::text[],'{-715827884,715827881,2147483647}'::text[],true) WHERE rows_written > 0
|
||||
NOTICE: executing the command locally: SELECT bytes FROM fetch_intermediate_results(ARRAY['repartitioned_results_xxxxx_from_4213582_to_0','repartitioned_results_xxxxx_from_4213584_to_0']::text[],'localhost',57638) bytes
|
||||
NOTICE: executing the command locally: INSERT INTO multi_mx_insert_select_repartition.target_table_4213585 AS citus_table_alias (a) SELECT a FROM read_intermediate_results('{repartitioned_results_xxxxx_from_4213581_to_0,repartitioned_results_xxxxx_from_4213582_to_0,repartitioned_results_xxxxx_from_4213584_to_0}'::text[], 'binary'::citus_copy_format) intermediate_result(a integer) RETURNING citus_table_alias.a
|
||||
NOTICE: executing the command locally: INSERT INTO multi_mx_insert_select_repartition.target_table_4213587 AS citus_table_alias (a) SELECT a FROM read_intermediate_results('{repartitioned_results_xxxxx_from_4213581_to_2}'::text[], 'binary'::citus_copy_format) intermediate_result(a integer) RETURNING citus_table_alias.a
|
||||
NOTICE: executing the command locally: INSERT INTO multi_mx_insert_select_repartition.target_table_4213585 AS citus_table_alias (a) SELECT intermediate_result.a FROM read_intermediate_results('{repartitioned_results_xxxxx_from_4213581_to_0,repartitioned_results_xxxxx_from_4213582_to_0,repartitioned_results_xxxxx_from_4213584_to_0}'::text[], 'binary'::citus_copy_format) intermediate_result(a integer) RETURNING citus_table_alias.a
|
||||
NOTICE: executing the command locally: INSERT INTO multi_mx_insert_select_repartition.target_table_4213587 AS citus_table_alias (a) SELECT intermediate_result.a FROM read_intermediate_results('{repartitioned_results_xxxxx_from_4213581_to_2}'::text[], 'binary'::citus_copy_format) intermediate_result(a integer) RETURNING citus_table_alias.a
|
||||
a
|
||||
---------------------------------------------------------------------
|
||||
0
|
||||
|
|
|
|||
|
|
@ -0,0 +1,160 @@
|
|||
--
|
||||
-- MULTI_MX_INSERT_SELECT_REPARTITION
|
||||
--
|
||||
-- Test queries on a distributed table with shards on the coordinator
|
||||
--
|
||||
-- This test file has an alternative output because of the change in the
|
||||
-- display of SQL-standard function's arguments in INSERT/SELECT in PG15.
|
||||
-- The alternative output can be deleted when we drop support for PG14
|
||||
--
|
||||
-- Test behaviour of repartitioned INSERT ... SELECT in MX setup
|
||||
CREATE SCHEMA multi_mx_insert_select_repartition;
|
||||
SET search_path TO multi_mx_insert_select_repartition;
|
||||
SET citus.next_shard_id TO 4213581;
|
||||
SET citus.shard_replication_factor TO 1;
|
||||
SET citus.shard_count TO 4;
|
||||
CREATE TABLE source_table(a int, b int);
|
||||
SELECT create_distributed_table('source_table', 'a');
|
||||
create_distributed_table
|
||||
---------------------------------------------------------------------
|
||||
|
||||
(1 row)
|
||||
|
||||
INSERT INTO source_table SELECT floor(i/4), i*i FROM generate_series(1, 20) i;
|
||||
SET citus.shard_count TO 3;
|
||||
CREATE TABLE target_table(a int, b int);
|
||||
SELECT create_distributed_table('target_table', 'a');
|
||||
create_distributed_table
|
||||
---------------------------------------------------------------------
|
||||
|
||||
(1 row)
|
||||
|
||||
CREATE FUNCTION square(int) RETURNS INT
|
||||
AS $$ SELECT $1 * $1 $$
|
||||
LANGUAGE SQL;
|
||||
select create_distributed_function('square(int)');
|
||||
NOTICE: procedure multi_mx_insert_select_repartition.square is already distributed
|
||||
DETAIL: Citus distributes procedures with CREATE [PROCEDURE|FUNCTION|AGGREGATE] commands
|
||||
create_distributed_function
|
||||
---------------------------------------------------------------------
|
||||
|
||||
(1 row)
|
||||
|
||||
select public.colocate_proc_with_table('square', 'source_table'::regclass, 0);
|
||||
colocate_proc_with_table
|
||||
---------------------------------------------------------------------
|
||||
|
||||
(1 row)
|
||||
|
||||
-- Test along with function delegation
|
||||
-- function delegation only happens for "SELECT f()", and we don't use
|
||||
-- repartitioned INSERT/SELECT when task count is 1, so the following
|
||||
-- should go via coordinator
|
||||
EXPLAIN (costs off) INSERT INTO target_table(a) SELECT square(4);
|
||||
QUERY PLAN
|
||||
---------------------------------------------------------------------
|
||||
Custom Scan (Citus INSERT ... SELECT)
|
||||
INSERT/SELECT method: pull to coordinator
|
||||
-> Result
|
||||
(3 rows)
|
||||
|
||||
INSERT INTO target_table(a) SELECT square(4);
|
||||
SELECT * FROM target_table;
|
||||
a | b
|
||||
---------------------------------------------------------------------
|
||||
16 |
|
||||
(1 row)
|
||||
|
||||
TRUNCATE target_table;
|
||||
--
|
||||
-- Test repartitioned INSERT/SELECT from MX worker
|
||||
--
|
||||
\c - - - :worker_1_port
|
||||
SET search_path TO multi_mx_insert_select_repartition;
|
||||
EXPLAIN (costs off) INSERT INTO target_table SELECT a, max(b) FROM source_table GROUP BY a;
|
||||
QUERY PLAN
|
||||
---------------------------------------------------------------------
|
||||
Custom Scan (Citus INSERT ... SELECT)
|
||||
INSERT/SELECT method: repartition
|
||||
-> Custom Scan (Citus Adaptive)
|
||||
Task Count: 4
|
||||
Tasks Shown: One of 4
|
||||
-> Task
|
||||
Node: host=localhost port=xxxxx dbname=regression
|
||||
-> HashAggregate
|
||||
Group Key: a
|
||||
-> Seq Scan on source_table_4213581 source_table
|
||||
(10 rows)
|
||||
|
||||
INSERT INTO target_table SELECT a, max(b) FROM source_table GROUP BY a;
|
||||
SET citus.log_local_commands to on;
|
||||
-- INSERT .. SELECT via repartitioning with local execution
|
||||
BEGIN;
|
||||
select count(*) from source_table WHERE a = 1;
|
||||
NOTICE: executing the command locally: SELECT count(*) AS count FROM multi_mx_insert_select_repartition.source_table_4213581 source_table WHERE (a OPERATOR(pg_catalog.=) 1)
|
||||
count
|
||||
---------------------------------------------------------------------
|
||||
4
|
||||
(1 row)
|
||||
|
||||
insert into target_table SELECT a*2 FROM source_table RETURNING a;
|
||||
NOTICE: executing the command locally: SELECT partition_index, 'repartitioned_results_xxxxx_from_4213581_to' || '_' || partition_index::text , rows_written FROM worker_partition_query_result('repartitioned_results_xxxxx_from_4213581_to','SELECT (a OPERATOR(pg_catalog.*) 2) AS a FROM multi_mx_insert_select_repartition.source_table_4213581 source_table WHERE true',0,'hash','{-2147483648,-715827883,715827882}'::text[],'{-715827884,715827881,2147483647}'::text[],true) WHERE rows_written > 0
|
||||
NOTICE: executing the command locally: SELECT partition_index, 'repartitioned_results_xxxxx_from_4213583_to' || '_' || partition_index::text , rows_written FROM worker_partition_query_result('repartitioned_results_xxxxx_from_4213583_to','SELECT (a OPERATOR(pg_catalog.*) 2) AS a FROM multi_mx_insert_select_repartition.source_table_4213583 source_table WHERE true',0,'hash','{-2147483648,-715827883,715827882}'::text[],'{-715827884,715827881,2147483647}'::text[],true) WHERE rows_written > 0
|
||||
NOTICE: executing the command locally: SELECT bytes FROM fetch_intermediate_results(ARRAY['repartitioned_results_xxxxx_from_4213582_to_0','repartitioned_results_xxxxx_from_4213584_to_0']::text[],'localhost',57638) bytes
|
||||
NOTICE: executing the command locally: INSERT INTO multi_mx_insert_select_repartition.target_table_4213585 AS citus_table_alias (a) SELECT a FROM read_intermediate_results('{repartitioned_results_xxxxx_from_4213581_to_0,repartitioned_results_xxxxx_from_4213582_to_0,repartitioned_results_xxxxx_from_4213584_to_0}'::text[], 'binary'::citus_copy_format) intermediate_result(a integer) RETURNING citus_table_alias.a
|
||||
NOTICE: executing the command locally: INSERT INTO multi_mx_insert_select_repartition.target_table_4213587 AS citus_table_alias (a) SELECT a FROM read_intermediate_results('{repartitioned_results_xxxxx_from_4213581_to_2}'::text[], 'binary'::citus_copy_format) intermediate_result(a integer) RETURNING citus_table_alias.a
|
||||
a
|
||||
---------------------------------------------------------------------
|
||||
0
|
||||
0
|
||||
0
|
||||
2
|
||||
2
|
||||
2
|
||||
2
|
||||
4
|
||||
4
|
||||
4
|
||||
4
|
||||
6
|
||||
6
|
||||
6
|
||||
6
|
||||
8
|
||||
8
|
||||
8
|
||||
8
|
||||
10
|
||||
(20 rows)
|
||||
|
||||
ROLLBACK;
|
||||
BEGIN;
|
||||
select count(*) from source_table WHERE a = 1;
|
||||
NOTICE: executing the command locally: SELECT count(*) AS count FROM multi_mx_insert_select_repartition.source_table_4213581 source_table WHERE (a OPERATOR(pg_catalog.=) 1)
|
||||
count
|
||||
---------------------------------------------------------------------
|
||||
4
|
||||
(1 row)
|
||||
|
||||
insert into target_table SELECT a FROM source_table LIMIT 10;
|
||||
NOTICE: executing the command locally: SELECT a FROM multi_mx_insert_select_repartition.source_table_4213581 source_table WHERE true LIMIT '10'::bigint
|
||||
NOTICE: executing the command locally: SELECT a FROM multi_mx_insert_select_repartition.source_table_4213583 source_table WHERE true LIMIT '10'::bigint
|
||||
NOTICE: executing the copy locally for shard xxxxx
|
||||
ROLLBACK;
|
||||
\c - - - :master_port
|
||||
SET search_path TO multi_mx_insert_select_repartition;
|
||||
SELECT * FROM target_table ORDER BY a;
|
||||
a | b
|
||||
---------------------------------------------------------------------
|
||||
0 | 9
|
||||
1 | 49
|
||||
2 | 121
|
||||
3 | 225
|
||||
4 | 361
|
||||
5 | 400
|
||||
(6 rows)
|
||||
|
||||
RESET client_min_messages;
|
||||
\set VERBOSITY terse
|
||||
DROP SCHEMA multi_mx_insert_select_repartition CASCADE;
|
||||
NOTICE: drop cascades to 3 other objects
|
||||
|
|
@ -1,3 +1,12 @@
|
|||
--
|
||||
-- MX_COORDINATOR_SHOULDHAVESHARDS
|
||||
--
|
||||
-- Test queries on a distributed table with shards on the coordinator
|
||||
--
|
||||
-- This test file has an alternative output because of the change in the
|
||||
-- display of SQL-standard function's arguments in INSERT/SELECT in PG15.
|
||||
-- The alternative output can be deleted when we drop support for PG14
|
||||
--
|
||||
CREATE SCHEMA mx_coordinator_shouldhaveshards;
|
||||
SET search_path TO mx_coordinator_shouldhaveshards;
|
||||
SET citus.shard_replication_factor to 1;
|
||||
|
|
@ -99,7 +108,7 @@ inserts AS (
|
|||
RETURNING *
|
||||
) SELECT count(*) FROM inserts;
|
||||
DEBUG: generating subplan XXX_1 for CTE stats: SELECT count(key) AS m FROM mx_coordinator_shouldhaveshards.table_1
|
||||
DEBUG: generating subplan XXX_2 for CTE inserts: INSERT INTO mx_coordinator_shouldhaveshards.table_2 (key, value) SELECT key, count(*) AS count FROM mx_coordinator_shouldhaveshards.table_1 WHERE (key OPERATOR(pg_catalog.>=) (SELECT stats.m FROM (SELECT intermediate_result.m FROM read_intermediate_result('XXX_1'::text, 'binary'::citus_copy_format) intermediate_result(m bigint)) stats)) GROUP BY key HAVING (count(*) OPERATOR(pg_catalog.<=) (SELECT stats.m FROM (SELECT intermediate_result.m FROM read_intermediate_result('XXX_1'::text, 'binary'::citus_copy_format) intermediate_result(m bigint)) stats)) LIMIT 1 RETURNING table_2.key, table_2.value
|
||||
DEBUG: generating subplan XXX_2 for CTE inserts: INSERT INTO mx_coordinator_shouldhaveshards.table_2 (key, value) SELECT table_1.key, count(*) AS count FROM mx_coordinator_shouldhaveshards.table_1 WHERE (table_1.key OPERATOR(pg_catalog.>=) (SELECT stats.m FROM (SELECT intermediate_result.m FROM read_intermediate_result('XXX_1'::text, 'binary'::citus_copy_format) intermediate_result(m bigint)) stats)) GROUP BY table_1.key HAVING (count(*) OPERATOR(pg_catalog.<=) (SELECT stats.m FROM (SELECT intermediate_result.m FROM read_intermediate_result('XXX_1'::text, 'binary'::citus_copy_format) intermediate_result(m bigint)) stats)) LIMIT 1 RETURNING table_2.key, table_2.value
|
||||
DEBUG: LIMIT clauses are not allowed in distributed INSERT ... SELECT queries
|
||||
DEBUG: push down of limit count: 1
|
||||
DEBUG: Plan XXX query after replacing subqueries and CTEs: SELECT count(*) AS count FROM (SELECT intermediate_result.key, intermediate_result.value FROM read_intermediate_result('XXX_2'::text, 'binary'::citus_copy_format) intermediate_result(key integer, value text)) inserts
|
||||
|
|
@ -160,7 +169,7 @@ inserts AS (
|
|||
RETURNING *
|
||||
) SELECT count(*) FROM inserts;
|
||||
DEBUG: generating subplan XXX_1 for CTE stats: SELECT count(key) AS m FROM mx_coordinator_shouldhaveshards.table_1_rep
|
||||
DEBUG: generating subplan XXX_2 for CTE inserts: INSERT INTO mx_coordinator_shouldhaveshards.table_2_rep (key, value) SELECT key, count(*) AS count FROM mx_coordinator_shouldhaveshards.table_1_rep WHERE (key OPERATOR(pg_catalog.>=) (SELECT stats.m FROM (SELECT intermediate_result.m FROM read_intermediate_result('XXX_1'::text, 'binary'::citus_copy_format) intermediate_result(m bigint)) stats)) GROUP BY key HAVING (count(*) OPERATOR(pg_catalog.<=) (SELECT stats.m FROM (SELECT intermediate_result.m FROM read_intermediate_result('XXX_1'::text, 'binary'::citus_copy_format) intermediate_result(m bigint)) stats)) LIMIT 1 RETURNING table_2_rep.key, table_2_rep.value
|
||||
DEBUG: generating subplan XXX_2 for CTE inserts: INSERT INTO mx_coordinator_shouldhaveshards.table_2_rep (key, value) SELECT table_1_rep.key, count(*) AS count FROM mx_coordinator_shouldhaveshards.table_1_rep WHERE (table_1_rep.key OPERATOR(pg_catalog.>=) (SELECT stats.m FROM (SELECT intermediate_result.m FROM read_intermediate_result('XXX_1'::text, 'binary'::citus_copy_format) intermediate_result(m bigint)) stats)) GROUP BY table_1_rep.key HAVING (count(*) OPERATOR(pg_catalog.<=) (SELECT stats.m FROM (SELECT intermediate_result.m FROM read_intermediate_result('XXX_1'::text, 'binary'::citus_copy_format) intermediate_result(m bigint)) stats)) LIMIT 1 RETURNING table_2_rep.key, table_2_rep.value
|
||||
DEBUG: LIMIT clauses are not allowed in distributed INSERT ... SELECT queries
|
||||
DEBUG: push down of limit count: 1
|
||||
DEBUG: Plan XXX query after replacing subqueries and CTEs: SELECT count(*) AS count FROM (SELECT intermediate_result.key, intermediate_result.value FROM read_intermediate_result('XXX_2'::text, 'binary'::citus_copy_format) intermediate_result(key integer, value text)) inserts
|
||||
|
|
@ -225,7 +234,7 @@ inserts AS (
|
|||
RETURNING *
|
||||
) SELECT count(*) FROM inserts;
|
||||
DEBUG: generating subplan XXX_1 for CTE stats: SELECT count(key) AS m FROM mx_coordinator_shouldhaveshards.table_1
|
||||
DEBUG: generating subplan XXX_2 for CTE inserts: INSERT INTO mx_coordinator_shouldhaveshards.table_2 (key, value) SELECT key, count(*) AS count FROM mx_coordinator_shouldhaveshards.table_1 WHERE (key OPERATOR(pg_catalog.>=) (SELECT stats.m FROM (SELECT intermediate_result.m FROM read_intermediate_result('XXX_1'::text, 'binary'::citus_copy_format) intermediate_result(m bigint)) stats)) GROUP BY key HAVING (count(*) OPERATOR(pg_catalog.<=) (SELECT stats.m FROM (SELECT intermediate_result.m FROM read_intermediate_result('XXX_1'::text, 'binary'::citus_copy_format) intermediate_result(m bigint)) stats)) LIMIT 1 RETURNING table_2.key, table_2.value
|
||||
DEBUG: generating subplan XXX_2 for CTE inserts: INSERT INTO mx_coordinator_shouldhaveshards.table_2 (key, value) SELECT table_1.key, count(*) AS count FROM mx_coordinator_shouldhaveshards.table_1 WHERE (table_1.key OPERATOR(pg_catalog.>=) (SELECT stats.m FROM (SELECT intermediate_result.m FROM read_intermediate_result('XXX_1'::text, 'binary'::citus_copy_format) intermediate_result(m bigint)) stats)) GROUP BY table_1.key HAVING (count(*) OPERATOR(pg_catalog.<=) (SELECT stats.m FROM (SELECT intermediate_result.m FROM read_intermediate_result('XXX_1'::text, 'binary'::citus_copy_format) intermediate_result(m bigint)) stats)) LIMIT 1 RETURNING table_2.key, table_2.value
|
||||
DEBUG: LIMIT clauses are not allowed in distributed INSERT ... SELECT queries
|
||||
DEBUG: push down of limit count: 1
|
||||
DEBUG: Plan XXX query after replacing subqueries and CTEs: SELECT count(*) AS count FROM (SELECT intermediate_result.key, intermediate_result.value FROM read_intermediate_result('XXX_2'::text, 'binary'::citus_copy_format) intermediate_result(key integer, value text)) inserts
|
||||
|
|
@ -286,7 +295,7 @@ inserts AS (
|
|||
RETURNING *
|
||||
) SELECT count(*) FROM inserts;
|
||||
DEBUG: generating subplan XXX_1 for CTE stats: SELECT count(key) AS m FROM mx_coordinator_shouldhaveshards.table_1_rep
|
||||
DEBUG: generating subplan XXX_2 for CTE inserts: INSERT INTO mx_coordinator_shouldhaveshards.table_2_rep (key, value) SELECT key, count(*) AS count FROM mx_coordinator_shouldhaveshards.table_1_rep WHERE (key OPERATOR(pg_catalog.>=) (SELECT stats.m FROM (SELECT intermediate_result.m FROM read_intermediate_result('XXX_1'::text, 'binary'::citus_copy_format) intermediate_result(m bigint)) stats)) GROUP BY key HAVING (count(*) OPERATOR(pg_catalog.<=) (SELECT stats.m FROM (SELECT intermediate_result.m FROM read_intermediate_result('XXX_1'::text, 'binary'::citus_copy_format) intermediate_result(m bigint)) stats)) LIMIT 1 RETURNING table_2_rep.key, table_2_rep.value
|
||||
DEBUG: generating subplan XXX_2 for CTE inserts: INSERT INTO mx_coordinator_shouldhaveshards.table_2_rep (key, value) SELECT table_1_rep.key, count(*) AS count FROM mx_coordinator_shouldhaveshards.table_1_rep WHERE (table_1_rep.key OPERATOR(pg_catalog.>=) (SELECT stats.m FROM (SELECT intermediate_result.m FROM read_intermediate_result('XXX_1'::text, 'binary'::citus_copy_format) intermediate_result(m bigint)) stats)) GROUP BY table_1_rep.key HAVING (count(*) OPERATOR(pg_catalog.<=) (SELECT stats.m FROM (SELECT intermediate_result.m FROM read_intermediate_result('XXX_1'::text, 'binary'::citus_copy_format) intermediate_result(m bigint)) stats)) LIMIT 1 RETURNING table_2_rep.key, table_2_rep.value
|
||||
DEBUG: LIMIT clauses are not allowed in distributed INSERT ... SELECT queries
|
||||
DEBUG: push down of limit count: 1
|
||||
DEBUG: Plan XXX query after replacing subqueries and CTEs: SELECT count(*) AS count FROM (SELECT intermediate_result.key, intermediate_result.value FROM read_intermediate_result('XXX_2'::text, 'binary'::citus_copy_format) intermediate_result(key integer, value text)) inserts
|
||||
|
|
|
|||
|
|
@ -0,0 +1,326 @@
|
|||
--
|
||||
-- MX_COORDINATOR_SHOULDHAVESHARDS
|
||||
--
|
||||
-- Test queries on a distributed table with shards on the coordinator
|
||||
--
|
||||
-- This test file has an alternative output because of the change in the
|
||||
-- display of SQL-standard function's arguments in INSERT/SELECT in PG15.
|
||||
-- The alternative output can be deleted when we drop support for PG14
|
||||
--
|
||||
CREATE SCHEMA mx_coordinator_shouldhaveshards;
|
||||
SET search_path TO mx_coordinator_shouldhaveshards;
|
||||
SET citus.shard_replication_factor to 1;
|
||||
SET client_min_messages TO WARNING;
|
||||
SELECT 1 FROM master_add_node('localhost', :master_port, groupid => 0);
|
||||
?column?
|
||||
---------------------------------------------------------------------
|
||||
1
|
||||
(1 row)
|
||||
|
||||
RESET client_min_messages;
|
||||
SELECT 1 FROM master_set_node_property('localhost', :master_port, 'shouldhaveshards', true);
|
||||
?column?
|
||||
---------------------------------------------------------------------
|
||||
1
|
||||
(1 row)
|
||||
|
||||
-- issue 4508 table_1 and table_2 are used to test some edge cases
|
||||
-- around intermediate result pruning
|
||||
CREATE TABLE table_1 (key int, value text);
|
||||
SELECT create_distributed_table('table_1', 'key', colocate_with := 'none');
|
||||
create_distributed_table
|
||||
---------------------------------------------------------------------
|
||||
|
||||
(1 row)
|
||||
|
||||
CREATE TABLE table_2 (key int, value text);
|
||||
SELECT create_distributed_table('table_2', 'key', colocate_with := 'none');
|
||||
create_distributed_table
|
||||
---------------------------------------------------------------------
|
||||
|
||||
(1 row)
|
||||
|
||||
INSERT INTO table_1 VALUES (1, '1'), (2, '2'), (3, '3'), (4, '4');
|
||||
INSERT INTO table_2 VALUES (1, '1'), (2, '2'), (3, '3'), (4, '4'), (5, '5'), (6, '6');
|
||||
SET citus.shard_replication_factor to 2;
|
||||
CREATE TABLE table_1_rep (key int, value text);
|
||||
SELECT create_distributed_table('table_1_rep', 'key', colocate_with := 'none');
|
||||
create_distributed_table
|
||||
---------------------------------------------------------------------
|
||||
|
||||
(1 row)
|
||||
|
||||
CREATE TABLE table_2_rep (key int, value text);
|
||||
SELECT create_distributed_table('table_2_rep', 'key', colocate_with := 'none');
|
||||
create_distributed_table
|
||||
---------------------------------------------------------------------
|
||||
|
||||
(1 row)
|
||||
|
||||
INSERT INTO table_1_rep VALUES (1, '1'), (2, '2'), (3, '3'), (4, '4');
|
||||
INSERT INTO table_2_rep VALUES (1, '1'), (2, '2'), (3, '3'), (4, '4'), (5, '5'), (6, '6');
|
||||
set citus.log_intermediate_results TO ON;
|
||||
set client_min_messages to debug1;
|
||||
WITH a AS (SELECT * FROM table_1 ORDER BY 1,2 DESC LIMIT 1)
|
||||
SELECT count(*),
|
||||
key
|
||||
FROM a JOIN table_2 USING (key)
|
||||
GROUP BY key
|
||||
HAVING (max(table_2.value) >= (SELECT value FROM a));
|
||||
DEBUG: generating subplan XXX_1 for CTE a: SELECT key, value FROM mx_coordinator_shouldhaveshards.table_1 ORDER BY key, value DESC LIMIT 1
|
||||
DEBUG: push down of limit count: 1
|
||||
DEBUG: Plan XXX query after replacing subqueries and CTEs: SELECT count(*) AS count, a.key FROM ((SELECT intermediate_result.key, intermediate_result.value FROM read_intermediate_result('XXX_1'::text, 'binary'::citus_copy_format) intermediate_result(key integer, value text)) a JOIN mx_coordinator_shouldhaveshards.table_2 USING (key)) GROUP BY a.key HAVING (max(table_2.value) OPERATOR(pg_catalog.>=) (SELECT a_1.value FROM (SELECT intermediate_result.key, intermediate_result.value FROM read_intermediate_result('XXX_1'::text, 'binary'::citus_copy_format) intermediate_result(key integer, value text)) a_1))
|
||||
DEBUG: Subplan XXX_1 will be written to local file
|
||||
DEBUG: Subplan XXX_1 will be sent to localhost:xxxxx
|
||||
DEBUG: Subplan XXX_1 will be sent to localhost:xxxxx
|
||||
count | key
|
||||
---------------------------------------------------------------------
|
||||
1 | 1
|
||||
(1 row)
|
||||
|
||||
WITH a AS (SELECT * FROM table_1 ORDER BY 1,2 DESC LIMIT 1)
|
||||
INSERT INTO table_1 SELECT count(*),
|
||||
key
|
||||
FROM a JOIN table_2 USING (key)
|
||||
GROUP BY key
|
||||
HAVING (max(table_2.value) >= (SELECT value FROM a));
|
||||
DEBUG: Group by list without distribution column is not allowed in distributed INSERT ... SELECT queries
|
||||
DEBUG: generating subplan XXX_1 for CTE a: SELECT key, value FROM mx_coordinator_shouldhaveshards.table_1 ORDER BY key, value DESC LIMIT 1
|
||||
DEBUG: push down of limit count: 1
|
||||
DEBUG: generating subplan XXX_2 for subquery SELECT int4(count(*)) AS auto_coerced_by_citus_0, (a.key)::text AS auto_coerced_by_citus_1 FROM ((SELECT intermediate_result.key, intermediate_result.value FROM read_intermediate_result('XXX_1'::text, 'binary'::citus_copy_format) intermediate_result(key integer, value text)) a JOIN mx_coordinator_shouldhaveshards.table_2 USING (key)) GROUP BY a.key HAVING (max(table_2.value) OPERATOR(pg_catalog.>=) (SELECT a_1.value FROM (SELECT intermediate_result.key, intermediate_result.value FROM read_intermediate_result('XXX_1'::text, 'binary'::citus_copy_format) intermediate_result(key integer, value text)) a_1))
|
||||
DEBUG: Plan XXX query after replacing subqueries and CTEs: SELECT auto_coerced_by_citus_0 AS key, auto_coerced_by_citus_1 AS value FROM (SELECT intermediate_result.auto_coerced_by_citus_0, intermediate_result.auto_coerced_by_citus_1 FROM read_intermediate_result('XXX_2'::text, 'binary'::citus_copy_format) intermediate_result(auto_coerced_by_citus_0 integer, auto_coerced_by_citus_1 text)) citus_insert_select_subquery
|
||||
DEBUG: Collecting INSERT ... SELECT results on coordinator
|
||||
DEBUG: Subplan XXX_1 will be written to local file
|
||||
DEBUG: Subplan XXX_1 will be sent to localhost:xxxxx
|
||||
DEBUG: Subplan XXX_1 will be sent to localhost:xxxxx
|
||||
DEBUG: Subplan XXX_2 will be written to local file
|
||||
WITH stats AS (
|
||||
SELECT count(key) m FROM table_1
|
||||
),
|
||||
inserts AS (
|
||||
INSERT INTO table_2
|
||||
SELECT key, count(*)
|
||||
FROM table_1
|
||||
WHERE key >= (SELECT m FROM stats)
|
||||
GROUP BY key
|
||||
HAVING count(*) <= (SELECT m FROM stats)
|
||||
LIMIT 1
|
||||
RETURNING *
|
||||
) SELECT count(*) FROM inserts;
|
||||
DEBUG: generating subplan XXX_1 for CTE stats: SELECT count(key) AS m FROM mx_coordinator_shouldhaveshards.table_1
|
||||
DEBUG: generating subplan XXX_2 for CTE inserts: INSERT INTO mx_coordinator_shouldhaveshards.table_2 (key, value) SELECT key, count(*) AS count FROM mx_coordinator_shouldhaveshards.table_1 WHERE (key OPERATOR(pg_catalog.>=) (SELECT stats.m FROM (SELECT intermediate_result.m FROM read_intermediate_result('XXX_1'::text, 'binary'::citus_copy_format) intermediate_result(m bigint)) stats)) GROUP BY key HAVING (count(*) OPERATOR(pg_catalog.<=) (SELECT stats.m FROM (SELECT intermediate_result.m FROM read_intermediate_result('XXX_1'::text, 'binary'::citus_copy_format) intermediate_result(m bigint)) stats)) LIMIT 1 RETURNING table_2.key, table_2.value
|
||||
DEBUG: LIMIT clauses are not allowed in distributed INSERT ... SELECT queries
|
||||
DEBUG: push down of limit count: 1
|
||||
DEBUG: Plan XXX query after replacing subqueries and CTEs: SELECT count(*) AS count FROM (SELECT intermediate_result.key, intermediate_result.value FROM read_intermediate_result('XXX_2'::text, 'binary'::citus_copy_format) intermediate_result(key integer, value text)) inserts
|
||||
DEBUG: Subplan XXX_1 will be written to local file
|
||||
DEBUG: Subplan XXX_1 will be sent to localhost:xxxxx
|
||||
DEBUG: Subplan XXX_1 will be sent to localhost:xxxxx
|
||||
DEBUG: Subplan XXX_2 will be written to local file
|
||||
DEBUG: Collecting INSERT ... SELECT results on coordinator
|
||||
count
|
||||
---------------------------------------------------------------------
|
||||
0
|
||||
(1 row)
|
||||
|
||||
WITH a AS (SELECT * FROM table_1_rep ORDER BY 1,2 DESC LIMIT 1)
|
||||
SELECT count(*),
|
||||
key
|
||||
FROM a JOIN table_2_rep USING (key)
|
||||
GROUP BY key
|
||||
HAVING (max(table_2_rep.value) >= (SELECT value FROM a));
|
||||
DEBUG: generating subplan XXX_1 for CTE a: SELECT key, value FROM mx_coordinator_shouldhaveshards.table_1_rep ORDER BY key, value DESC LIMIT 1
|
||||
DEBUG: push down of limit count: 1
|
||||
DEBUG: Plan XXX query after replacing subqueries and CTEs: SELECT count(*) AS count, a.key FROM ((SELECT intermediate_result.key, intermediate_result.value FROM read_intermediate_result('XXX_1'::text, 'binary'::citus_copy_format) intermediate_result(key integer, value text)) a JOIN mx_coordinator_shouldhaveshards.table_2_rep USING (key)) GROUP BY a.key HAVING (max(table_2_rep.value) OPERATOR(pg_catalog.>=) (SELECT a_1.value FROM (SELECT intermediate_result.key, intermediate_result.value FROM read_intermediate_result('XXX_1'::text, 'binary'::citus_copy_format) intermediate_result(key integer, value text)) a_1))
|
||||
DEBUG: Subplan XXX_1 will be written to local file
|
||||
DEBUG: Subplan XXX_1 will be sent to localhost:xxxxx
|
||||
DEBUG: Subplan XXX_1 will be sent to localhost:xxxxx
|
||||
count | key
|
||||
---------------------------------------------------------------------
|
||||
1 | 1
|
||||
(1 row)
|
||||
|
||||
WITH a AS (SELECT * FROM table_1_rep ORDER BY 1,2 DESC LIMIT 1)
|
||||
INSERT INTO table_1_rep SELECT count(*),
|
||||
key
|
||||
FROM a JOIN table_2_rep USING (key)
|
||||
GROUP BY key
|
||||
HAVING (max(table_2_rep.value) >= (SELECT value FROM a));
|
||||
DEBUG: Group by list without distribution column is not allowed in distributed INSERT ... SELECT queries
|
||||
DEBUG: generating subplan XXX_1 for CTE a: SELECT key, value FROM mx_coordinator_shouldhaveshards.table_1_rep ORDER BY key, value DESC LIMIT 1
|
||||
DEBUG: push down of limit count: 1
|
||||
DEBUG: generating subplan XXX_2 for subquery SELECT int4(count(*)) AS auto_coerced_by_citus_0, (a.key)::text AS auto_coerced_by_citus_1 FROM ((SELECT intermediate_result.key, intermediate_result.value FROM read_intermediate_result('XXX_1'::text, 'binary'::citus_copy_format) intermediate_result(key integer, value text)) a JOIN mx_coordinator_shouldhaveshards.table_2_rep USING (key)) GROUP BY a.key HAVING (max(table_2_rep.value) OPERATOR(pg_catalog.>=) (SELECT a_1.value FROM (SELECT intermediate_result.key, intermediate_result.value FROM read_intermediate_result('XXX_1'::text, 'binary'::citus_copy_format) intermediate_result(key integer, value text)) a_1))
|
||||
DEBUG: Plan XXX query after replacing subqueries and CTEs: SELECT auto_coerced_by_citus_0 AS key, auto_coerced_by_citus_1 AS value FROM (SELECT intermediate_result.auto_coerced_by_citus_0, intermediate_result.auto_coerced_by_citus_1 FROM read_intermediate_result('XXX_2'::text, 'binary'::citus_copy_format) intermediate_result(auto_coerced_by_citus_0 integer, auto_coerced_by_citus_1 text)) citus_insert_select_subquery
|
||||
DEBUG: Collecting INSERT ... SELECT results on coordinator
|
||||
DEBUG: Subplan XXX_1 will be written to local file
|
||||
DEBUG: Subplan XXX_1 will be sent to localhost:xxxxx
|
||||
DEBUG: Subplan XXX_1 will be sent to localhost:xxxxx
|
||||
DEBUG: Subplan XXX_2 will be written to local file
|
||||
WITH stats AS (
|
||||
SELECT count(key) m FROM table_1_rep
|
||||
),
|
||||
inserts AS (
|
||||
INSERT INTO table_2_rep
|
||||
SELECT key, count(*)
|
||||
FROM table_1_rep
|
||||
WHERE key >= (SELECT m FROM stats)
|
||||
GROUP BY key
|
||||
HAVING count(*) <= (SELECT m FROM stats)
|
||||
LIMIT 1
|
||||
RETURNING *
|
||||
) SELECT count(*) FROM inserts;
|
||||
DEBUG: generating subplan XXX_1 for CTE stats: SELECT count(key) AS m FROM mx_coordinator_shouldhaveshards.table_1_rep
|
||||
DEBUG: generating subplan XXX_2 for CTE inserts: INSERT INTO mx_coordinator_shouldhaveshards.table_2_rep (key, value) SELECT key, count(*) AS count FROM mx_coordinator_shouldhaveshards.table_1_rep WHERE (key OPERATOR(pg_catalog.>=) (SELECT stats.m FROM (SELECT intermediate_result.m FROM read_intermediate_result('XXX_1'::text, 'binary'::citus_copy_format) intermediate_result(m bigint)) stats)) GROUP BY key HAVING (count(*) OPERATOR(pg_catalog.<=) (SELECT stats.m FROM (SELECT intermediate_result.m FROM read_intermediate_result('XXX_1'::text, 'binary'::citus_copy_format) intermediate_result(m bigint)) stats)) LIMIT 1 RETURNING table_2_rep.key, table_2_rep.value
|
||||
DEBUG: LIMIT clauses are not allowed in distributed INSERT ... SELECT queries
|
||||
DEBUG: push down of limit count: 1
|
||||
DEBUG: Plan XXX query after replacing subqueries and CTEs: SELECT count(*) AS count FROM (SELECT intermediate_result.key, intermediate_result.value FROM read_intermediate_result('XXX_2'::text, 'binary'::citus_copy_format) intermediate_result(key integer, value text)) inserts
|
||||
DEBUG: Subplan XXX_1 will be written to local file
|
||||
DEBUG: Subplan XXX_1 will be sent to localhost:xxxxx
|
||||
DEBUG: Subplan XXX_1 will be sent to localhost:xxxxx
|
||||
DEBUG: Subplan XXX_2 will be written to local file
|
||||
DEBUG: Collecting INSERT ... SELECT results on coordinator
|
||||
count
|
||||
---------------------------------------------------------------------
|
||||
0
|
||||
(1 row)
|
||||
|
||||
\c - - - :worker_1_port
|
||||
SET search_path TO mx_coordinator_shouldhaveshards;
|
||||
set citus.log_intermediate_results TO ON;
|
||||
set client_min_messages to debug1;
|
||||
WITH a AS (SELECT * FROM table_1 ORDER BY 1,2 DESC LIMIT 1)
|
||||
SELECT count(*),
|
||||
key
|
||||
FROM a JOIN table_2 USING (key)
|
||||
GROUP BY key
|
||||
HAVING (max(table_2.value) >= (SELECT value FROM a));
|
||||
DEBUG: generating subplan XXX_1 for CTE a: SELECT key, value FROM mx_coordinator_shouldhaveshards.table_1 ORDER BY key, value DESC LIMIT 1
|
||||
DEBUG: push down of limit count: 1
|
||||
DEBUG: Plan XXX query after replacing subqueries and CTEs: SELECT count(*) AS count, a.key FROM ((SELECT intermediate_result.key, intermediate_result.value FROM read_intermediate_result('XXX_1'::text, 'binary'::citus_copy_format) intermediate_result(key integer, value text)) a JOIN mx_coordinator_shouldhaveshards.table_2 USING (key)) GROUP BY a.key HAVING (max(table_2.value) OPERATOR(pg_catalog.>=) (SELECT a_1.value FROM (SELECT intermediate_result.key, intermediate_result.value FROM read_intermediate_result('XXX_1'::text, 'binary'::citus_copy_format) intermediate_result(key integer, value text)) a_1))
|
||||
DEBUG: Subplan XXX_1 will be written to local file
|
||||
DEBUG: Subplan XXX_1 will be sent to localhost:xxxxx
|
||||
DEBUG: Subplan XXX_1 will be sent to localhost:xxxxx
|
||||
count | key
|
||||
---------------------------------------------------------------------
|
||||
1 | 1
|
||||
(1 row)
|
||||
|
||||
WITH a AS (SELECT * FROM table_1 ORDER BY 1,2 DESC LIMIT 1)
|
||||
INSERT INTO table_1 SELECT count(*),
|
||||
key
|
||||
FROM a JOIN table_2 USING (key)
|
||||
GROUP BY key
|
||||
HAVING (max(table_2.value) >= (SELECT value FROM a));
|
||||
DEBUG: Group by list without distribution column is not allowed in distributed INSERT ... SELECT queries
|
||||
DEBUG: generating subplan XXX_1 for CTE a: SELECT key, value FROM mx_coordinator_shouldhaveshards.table_1 ORDER BY key, value DESC LIMIT 1
|
||||
DEBUG: push down of limit count: 1
|
||||
DEBUG: generating subplan XXX_2 for subquery SELECT int4(count(*)) AS auto_coerced_by_citus_0, (a.key)::text AS auto_coerced_by_citus_1 FROM ((SELECT intermediate_result.key, intermediate_result.value FROM read_intermediate_result('XXX_1'::text, 'binary'::citus_copy_format) intermediate_result(key integer, value text)) a JOIN mx_coordinator_shouldhaveshards.table_2 USING (key)) GROUP BY a.key HAVING (max(table_2.value) OPERATOR(pg_catalog.>=) (SELECT a_1.value FROM (SELECT intermediate_result.key, intermediate_result.value FROM read_intermediate_result('XXX_1'::text, 'binary'::citus_copy_format) intermediate_result(key integer, value text)) a_1))
|
||||
DEBUG: Plan XXX query after replacing subqueries and CTEs: SELECT auto_coerced_by_citus_0 AS key, auto_coerced_by_citus_1 AS value FROM (SELECT intermediate_result.auto_coerced_by_citus_0, intermediate_result.auto_coerced_by_citus_1 FROM read_intermediate_result('XXX_2'::text, 'binary'::citus_copy_format) intermediate_result(auto_coerced_by_citus_0 integer, auto_coerced_by_citus_1 text)) citus_insert_select_subquery
|
||||
DEBUG: Collecting INSERT ... SELECT results on coordinator
|
||||
DEBUG: Subplan XXX_1 will be written to local file
|
||||
DEBUG: Subplan XXX_1 will be sent to localhost:xxxxx
|
||||
DEBUG: Subplan XXX_1 will be sent to localhost:xxxxx
|
||||
DEBUG: Subplan XXX_2 will be written to local file
|
||||
WITH stats AS (
|
||||
SELECT count(key) m FROM table_1
|
||||
),
|
||||
inserts AS (
|
||||
INSERT INTO table_2
|
||||
SELECT key, count(*)
|
||||
FROM table_1
|
||||
WHERE key >= (SELECT m FROM stats)
|
||||
GROUP BY key
|
||||
HAVING count(*) <= (SELECT m FROM stats)
|
||||
LIMIT 1
|
||||
RETURNING *
|
||||
) SELECT count(*) FROM inserts;
|
||||
DEBUG: generating subplan XXX_1 for CTE stats: SELECT count(key) AS m FROM mx_coordinator_shouldhaveshards.table_1
|
||||
DEBUG: generating subplan XXX_2 for CTE inserts: INSERT INTO mx_coordinator_shouldhaveshards.table_2 (key, value) SELECT key, count(*) AS count FROM mx_coordinator_shouldhaveshards.table_1 WHERE (key OPERATOR(pg_catalog.>=) (SELECT stats.m FROM (SELECT intermediate_result.m FROM read_intermediate_result('XXX_1'::text, 'binary'::citus_copy_format) intermediate_result(m bigint)) stats)) GROUP BY key HAVING (count(*) OPERATOR(pg_catalog.<=) (SELECT stats.m FROM (SELECT intermediate_result.m FROM read_intermediate_result('XXX_1'::text, 'binary'::citus_copy_format) intermediate_result(m bigint)) stats)) LIMIT 1 RETURNING table_2.key, table_2.value
|
||||
DEBUG: LIMIT clauses are not allowed in distributed INSERT ... SELECT queries
|
||||
DEBUG: push down of limit count: 1
|
||||
DEBUG: Plan XXX query after replacing subqueries and CTEs: SELECT count(*) AS count FROM (SELECT intermediate_result.key, intermediate_result.value FROM read_intermediate_result('XXX_2'::text, 'binary'::citus_copy_format) intermediate_result(key integer, value text)) inserts
|
||||
DEBUG: Subplan XXX_1 will be written to local file
|
||||
DEBUG: Subplan XXX_1 will be sent to localhost:xxxxx
|
||||
DEBUG: Subplan XXX_1 will be sent to localhost:xxxxx
|
||||
DEBUG: Subplan XXX_2 will be written to local file
|
||||
DEBUG: Collecting INSERT ... SELECT results on coordinator
|
||||
count
|
||||
---------------------------------------------------------------------
|
||||
0
|
||||
(1 row)
|
||||
|
||||
WITH a AS (SELECT * FROM table_1_rep ORDER BY 1,2 DESC LIMIT 1)
|
||||
SELECT count(*),
|
||||
key
|
||||
FROM a JOIN table_2_rep USING (key)
|
||||
GROUP BY key
|
||||
HAVING (max(table_2_rep.value) >= (SELECT value FROM a));
|
||||
DEBUG: generating subplan XXX_1 for CTE a: SELECT key, value FROM mx_coordinator_shouldhaveshards.table_1_rep ORDER BY key, value DESC LIMIT 1
|
||||
DEBUG: push down of limit count: 1
|
||||
DEBUG: Plan XXX query after replacing subqueries and CTEs: SELECT count(*) AS count, a.key FROM ((SELECT intermediate_result.key, intermediate_result.value FROM read_intermediate_result('XXX_1'::text, 'binary'::citus_copy_format) intermediate_result(key integer, value text)) a JOIN mx_coordinator_shouldhaveshards.table_2_rep USING (key)) GROUP BY a.key HAVING (max(table_2_rep.value) OPERATOR(pg_catalog.>=) (SELECT a_1.value FROM (SELECT intermediate_result.key, intermediate_result.value FROM read_intermediate_result('XXX_1'::text, 'binary'::citus_copy_format) intermediate_result(key integer, value text)) a_1))
|
||||
DEBUG: Subplan XXX_1 will be written to local file
|
||||
DEBUG: Subplan XXX_1 will be sent to localhost:xxxxx
|
||||
DEBUG: Subplan XXX_1 will be sent to localhost:xxxxx
|
||||
count | key
|
||||
---------------------------------------------------------------------
|
||||
1 | 1
|
||||
(1 row)
|
||||
|
||||
WITH a AS (SELECT * FROM table_1_rep ORDER BY 1,2 DESC LIMIT 1)
|
||||
INSERT INTO table_1_rep SELECT count(*),
|
||||
key
|
||||
FROM a JOIN table_2_rep USING (key)
|
||||
GROUP BY key
|
||||
HAVING (max(table_2_rep.value) >= (SELECT value FROM a));
|
||||
DEBUG: Group by list without distribution column is not allowed in distributed INSERT ... SELECT queries
|
||||
DEBUG: generating subplan XXX_1 for CTE a: SELECT key, value FROM mx_coordinator_shouldhaveshards.table_1_rep ORDER BY key, value DESC LIMIT 1
|
||||
DEBUG: push down of limit count: 1
|
||||
DEBUG: generating subplan XXX_2 for subquery SELECT int4(count(*)) AS auto_coerced_by_citus_0, (a.key)::text AS auto_coerced_by_citus_1 FROM ((SELECT intermediate_result.key, intermediate_result.value FROM read_intermediate_result('XXX_1'::text, 'binary'::citus_copy_format) intermediate_result(key integer, value text)) a JOIN mx_coordinator_shouldhaveshards.table_2_rep USING (key)) GROUP BY a.key HAVING (max(table_2_rep.value) OPERATOR(pg_catalog.>=) (SELECT a_1.value FROM (SELECT intermediate_result.key, intermediate_result.value FROM read_intermediate_result('XXX_1'::text, 'binary'::citus_copy_format) intermediate_result(key integer, value text)) a_1))
|
||||
DEBUG: Plan XXX query after replacing subqueries and CTEs: SELECT auto_coerced_by_citus_0 AS key, auto_coerced_by_citus_1 AS value FROM (SELECT intermediate_result.auto_coerced_by_citus_0, intermediate_result.auto_coerced_by_citus_1 FROM read_intermediate_result('XXX_2'::text, 'binary'::citus_copy_format) intermediate_result(auto_coerced_by_citus_0 integer, auto_coerced_by_citus_1 text)) citus_insert_select_subquery
|
||||
DEBUG: Collecting INSERT ... SELECT results on coordinator
|
||||
DEBUG: Subplan XXX_1 will be written to local file
|
||||
DEBUG: Subplan XXX_1 will be sent to localhost:xxxxx
|
||||
DEBUG: Subplan XXX_1 will be sent to localhost:xxxxx
|
||||
DEBUG: Subplan XXX_2 will be written to local file
|
||||
WITH stats AS (
|
||||
SELECT count(key) m FROM table_1_rep
|
||||
),
|
||||
inserts AS (
|
||||
INSERT INTO table_2_rep
|
||||
SELECT key, count(*)
|
||||
FROM table_1_rep
|
||||
WHERE key >= (SELECT m FROM stats)
|
||||
GROUP BY key
|
||||
HAVING count(*) <= (SELECT m FROM stats)
|
||||
LIMIT 1
|
||||
RETURNING *
|
||||
) SELECT count(*) FROM inserts;
|
||||
DEBUG: generating subplan XXX_1 for CTE stats: SELECT count(key) AS m FROM mx_coordinator_shouldhaveshards.table_1_rep
|
||||
DEBUG: generating subplan XXX_2 for CTE inserts: INSERT INTO mx_coordinator_shouldhaveshards.table_2_rep (key, value) SELECT key, count(*) AS count FROM mx_coordinator_shouldhaveshards.table_1_rep WHERE (key OPERATOR(pg_catalog.>=) (SELECT stats.m FROM (SELECT intermediate_result.m FROM read_intermediate_result('XXX_1'::text, 'binary'::citus_copy_format) intermediate_result(m bigint)) stats)) GROUP BY key HAVING (count(*) OPERATOR(pg_catalog.<=) (SELECT stats.m FROM (SELECT intermediate_result.m FROM read_intermediate_result('XXX_1'::text, 'binary'::citus_copy_format) intermediate_result(m bigint)) stats)) LIMIT 1 RETURNING table_2_rep.key, table_2_rep.value
|
||||
DEBUG: LIMIT clauses are not allowed in distributed INSERT ... SELECT queries
|
||||
DEBUG: push down of limit count: 1
|
||||
DEBUG: Plan XXX query after replacing subqueries and CTEs: SELECT count(*) AS count FROM (SELECT intermediate_result.key, intermediate_result.value FROM read_intermediate_result('XXX_2'::text, 'binary'::citus_copy_format) intermediate_result(key integer, value text)) inserts
|
||||
DEBUG: Subplan XXX_1 will be written to local file
|
||||
DEBUG: Subplan XXX_1 will be sent to localhost:xxxxx
|
||||
DEBUG: Subplan XXX_1 will be sent to localhost:xxxxx
|
||||
DEBUG: Subplan XXX_2 will be written to local file
|
||||
DEBUG: Collecting INSERT ... SELECT results on coordinator
|
||||
count
|
||||
---------------------------------------------------------------------
|
||||
0
|
||||
(1 row)
|
||||
|
||||
\c - - - :master_port
|
||||
SELECT 1 FROM master_set_node_property('localhost', :master_port, 'shouldhaveshards', false);
|
||||
?column?
|
||||
---------------------------------------------------------------------
|
||||
1
|
||||
(1 row)
|
||||
|
||||
SET client_min_messages TO ERROR;
|
||||
DROP SCHEMA mx_coordinator_shouldhaveshards CASCADE;
|
||||
SELECT master_remove_node('localhost', :master_port);
|
||||
master_remove_node
|
||||
---------------------------------------------------------------------
|
||||
|
||||
(1 row)
|
||||
|
||||
|
|
@ -1,3 +1,12 @@
|
|||
--
|
||||
-- CITUS_LOCAL_TABLES_QUERIES
|
||||
--
|
||||
-- Test queries on a distributed table with shards on the coordinator
|
||||
--
|
||||
-- This test file has an alternative output because of the change in the
|
||||
-- display of SQL-standard function's arguments in INSERT/SELECT in PG15.
|
||||
-- The alternative output can be deleted when we drop support for PG14
|
||||
--
|
||||
\set VERBOSITY terse
|
||||
|
||||
SET citus.next_shard_id TO 1509000;
|
||||
|
|
|
|||
|
|
@ -1,3 +1,12 @@
|
|||
--
|
||||
-- CTE_INLINE
|
||||
--
|
||||
-- Test queries on a distributed table with shards on the coordinator
|
||||
--
|
||||
-- This test file has an alternative output because of the change in the
|
||||
-- display of SQL-standard function's arguments in INSERT/SELECT in PG15.
|
||||
-- The alternative output can be deleted when we drop support for PG14
|
||||
--
|
||||
CREATE SCHEMA cte_inline;
|
||||
SET search_path TO cte_inline;
|
||||
SET citus.next_shard_id TO 1960000;
|
||||
|
|
|
|||
|
|
@ -1,3 +1,12 @@
|
|||
--
|
||||
-- INSERT_SELECT_REPARTITION
|
||||
--
|
||||
-- Test queries on a distributed table with shards on the coordinator
|
||||
--
|
||||
-- This test file has an alternative output because of the change in the
|
||||
-- display of SQL-standard function's arguments in INSERT/SELECT in PG15.
|
||||
-- The alternative output can be deleted when we drop support for PG14
|
||||
--
|
||||
-- tests behaviour of INSERT INTO ... SELECT with repartitioning
|
||||
CREATE SCHEMA insert_select_repartition;
|
||||
SET search_path TO 'insert_select_repartition';
|
||||
|
|
|
|||
|
|
@ -1,3 +1,13 @@
|
|||
--
|
||||
-- INTERMEDIATE_RESULT_PRUNING
|
||||
--
|
||||
-- Test queries on a distributed table with shards on the coordinator
|
||||
--
|
||||
-- This test file has an alternative output because of the change in the
|
||||
-- display of SQL-standard function's arguments in INSERT/SELECT in PG15.
|
||||
-- The alternative output can be deleted when we drop support for PG14
|
||||
--
|
||||
|
||||
CREATE SCHEMA intermediate_result_pruning;
|
||||
SET search_path TO intermediate_result_pruning;
|
||||
SET citus.log_intermediate_results TO TRUE;
|
||||
|
|
|
|||
|
|
@ -1,3 +1,12 @@
|
|||
--
|
||||
-- LOCAL_SHARD_EXECUTION
|
||||
--
|
||||
-- Test queries on a distributed table with shards on the coordinator
|
||||
--
|
||||
-- This test file has an alternative output because of the change in the
|
||||
-- display of SQL-standard function's arguments in INSERT/SELECT in PG15.
|
||||
-- The alternative output can be deleted when we drop support for PG14
|
||||
--
|
||||
CREATE SCHEMA local_shard_execution;
|
||||
SET search_path TO local_shard_execution;
|
||||
|
||||
|
|
|
|||
|
|
@ -1,3 +1,12 @@
|
|||
--
|
||||
-- LOCAL_SHARD_EXECUTION_REPLICATED
|
||||
--
|
||||
-- Test queries on a distributed table with shards on the coordinator
|
||||
--
|
||||
-- This test file has an alternative output because of the change in the
|
||||
-- display of SQL-standard function's arguments in INSERT/SELECT in PG15.
|
||||
-- The alternative output can be deleted when we drop support for PG14
|
||||
--
|
||||
CREATE SCHEMA local_shard_execution_replicated;
|
||||
SET search_path TO local_shard_execution_replicated;
|
||||
|
||||
|
|
|
|||
|
|
@ -1,3 +1,12 @@
|
|||
--
|
||||
-- MULTI_MX_INSERT_SELECT_REPARTITION
|
||||
--
|
||||
-- Test queries on a distributed table with shards on the coordinator
|
||||
--
|
||||
-- This test file has an alternative output because of the change in the
|
||||
-- display of SQL-standard function's arguments in INSERT/SELECT in PG15.
|
||||
-- The alternative output can be deleted when we drop support for PG14
|
||||
--
|
||||
-- Test behaviour of repartitioned INSERT ... SELECT in MX setup
|
||||
|
||||
CREATE SCHEMA multi_mx_insert_select_repartition;
|
||||
|
|
|
|||
|
|
@ -1,3 +1,12 @@
|
|||
--
|
||||
-- MX_COORDINATOR_SHOULDHAVESHARDS
|
||||
--
|
||||
-- Test queries on a distributed table with shards on the coordinator
|
||||
--
|
||||
-- This test file has an alternative output because of the change in the
|
||||
-- display of SQL-standard function's arguments in INSERT/SELECT in PG15.
|
||||
-- The alternative output can be deleted when we drop support for PG14
|
||||
--
|
||||
CREATE SCHEMA mx_coordinator_shouldhaveshards;
|
||||
SET search_path TO mx_coordinator_shouldhaveshards;
|
||||
|
||||
|
|
|
|||
Loading…
Reference in New Issue