\set VERBOSITY terse SET citus.next_shard_id TO 1510000; SET citus.shard_replication_factor TO 1; SET citus.enable_local_execution TO ON; SET citus.log_local_commands TO ON; CREATE SCHEMA citus_local_table_queries_mx; SET search_path TO citus_local_table_queries_mx; -- ensure that coordinator is added to pg_dist_node SET client_min_messages to ERROR; SELECT 1 FROM master_add_node('localhost', :master_port, groupId => 0); ?column? --------------------------------------------------------------------- 1 (1 row) RESET client_min_messages; -- start metadata sync to worker 1 SELECT start_metadata_sync_to_node('localhost', :worker_1_port); start_metadata_sync_to_node --------------------------------------------------------------------- (1 row) SET citus.shard_replication_factor TO 1; CREATE TABLE dummy_reference_table(a int unique, b int); SELECT create_reference_table('dummy_reference_table'); create_reference_table --------------------------------------------------------------------- (1 row) CREATE TABLE citus_local_table(a int, b int); ALTER TABLE citus_local_table ADD CONSTRAINT fkey_to_dummy_1 FOREIGN KEY (a) REFERENCES dummy_reference_table(a); NOTICE: executing the command locally: SELECT worker_apply_inter_shard_ddl_command (1510001, 'citus_local_table_queries_mx', 1510000, 'citus_local_table_queries_mx', 'ALTER TABLE citus_local_table ADD CONSTRAINT fkey_to_dummy_1 FOREIGN KEY (a) REFERENCES dummy_reference_table(a);') CREATE TABLE citus_local_table_2(a int, b int); ALTER TABLE citus_local_table_2 ADD CONSTRAINT fkey_to_dummy_2 FOREIGN KEY (a) REFERENCES dummy_reference_table(a); NOTICE: executing the command locally: SELECT worker_apply_inter_shard_ddl_command (1510002, 'citus_local_table_queries_mx', 1510000, 'citus_local_table_queries_mx', 'ALTER TABLE citus_local_table_2 ADD CONSTRAINT fkey_to_dummy_2 FOREIGN KEY (a) REFERENCES dummy_reference_table(a);') CREATE TABLE reference_table(a int, b int); SELECT create_reference_table('reference_table'); create_reference_table --------------------------------------------------------------------- (1 row) CREATE TABLE distributed_table(a int, b int); SELECT create_distributed_table('distributed_table', 'a'); create_distributed_table --------------------------------------------------------------------- (1 row) \c - - - :worker_1_port SET search_path TO citus_local_table_queries_mx; CREATE TABLE postgres_local_table(a int, b int); -- Define a helper function to truncate & insert some data into our test tables -- We should call this function at some places in this test file to prevent -- test to take a long time. -- We shouldn't use LIMIT in INSERT SELECT queries to make the test faster as -- LIMIT would force planner to wrap SELECT query in an intermediate result and -- this might reduce the coverage of the test cases. SET citus.enable_metadata_sync TO OFF; CREATE FUNCTION clear_and_init_test_tables() RETURNS void AS $$ BEGIN SET client_min_messages to ERROR; TRUNCATE postgres_local_table, citus_local_table, reference_table, distributed_table, dummy_reference_table, citus_local_table_2; INSERT INTO dummy_reference_table SELECT i, i FROM generate_series(0, 5) i; INSERT INTO citus_local_table SELECT i, i FROM generate_series(0, 5) i; INSERT INTO citus_local_table_2 SELECT i, i FROM generate_series(0, 5) i; INSERT INTO postgres_local_table SELECT i, i FROM generate_series(0, 5) i; INSERT INTO distributed_table SELECT i, i FROM generate_series(0, 5) i; INSERT INTO reference_table SELECT i, i FROM generate_series(0, 5) i; RESET client_min_messages; END; $$ LANGUAGE plpgsql; RESET citus.enable_metadata_sync; --------------------------------------------------------------------- ---- SELECT ---- --------------------------------------------------------------------- SELECT clear_and_init_test_tables(); clear_and_init_test_tables --------------------------------------------------------------------- (1 row) -- join between citus local tables and reference tables would succeed SELECT count(*) FROM citus_local_table, reference_table WHERE citus_local_table.a = reference_table.a; count --------------------------------------------------------------------- 6 (1 row) SELECT * FROM citus_local_table, reference_table WHERE citus_local_table.a = reference_table.a ORDER BY 1,2,3,4 FOR UPDATE; a | b | a | b --------------------------------------------------------------------- 0 | 0 | 0 | 0 1 | 1 | 1 | 1 2 | 2 | 2 | 2 3 | 3 | 3 | 3 4 | 4 | 4 | 4 5 | 5 | 5 | 5 (6 rows) -- should work WITH cte_1 AS (SELECT * FROM citus_local_table, reference_table WHERE citus_local_table.a = reference_table.a ORDER BY 1,2,3,4 FOR UPDATE) SELECT count(*) FROM cte_1; count --------------------------------------------------------------------- 6 (1 row) -- should work as joins are between ctes WITH cte_citus_local_table AS (SELECT * FROM citus_local_table), cte_postgres_local_table AS (SELECT * FROM postgres_local_table), cte_distributed_table AS (SELECT * FROM distributed_table) SELECT count(*) FROM cte_distributed_table, cte_citus_local_table, cte_postgres_local_table WHERE cte_citus_local_table.a = 1 AND cte_distributed_table.a = 1; count --------------------------------------------------------------------- 6 (1 row) -- should fail as we don't support direct joins between distributed/local tables SELECT count(*) FROM distributed_table d1, distributed_table d2, citus_local_table; ERROR: complex joins are only supported when all distributed tables are co-located and joined on their distribution columns -- local table inside subquery should just work SELECT count(*) FROM ( SELECT * FROM (SELECT * FROM citus_local_table) as subquery_inner ) as subquery_top; count --------------------------------------------------------------------- 6 (1 row) SELECT clear_and_init_test_tables(); clear_and_init_test_tables --------------------------------------------------------------------- (1 row) -- join between citus/postgres local tables wouldn't work as citus local table is on the coordinator SELECT count(*) FROM ( SELECT * FROM (SELECT count(*) FROM citus_local_table, postgres_local_table) as subquery_inner ) as subquery_top; ERROR: direct joins between distributed and local tables are not supported -- should fail as we don't support direct joins between distributed/local tables SELECT count(*) FROM ( SELECT *, random() FROM (SELECT *, random() FROM citus_local_table, distributed_table) as subquery_inner ) as subquery_top; count --------------------------------------------------------------------- 36 (1 row) -- should fail as we don't support direct joins between distributed/local tables SELECT count(*) FROM ( SELECT *, random() FROM ( WITH cte_1 AS (SELECT *, random() FROM citus_local_table, distributed_table) SELECT * FROM cte_1) as subquery_inner ) as subquery_top; count --------------------------------------------------------------------- 36 (1 row) -- should be fine SELECT count(*) FROM ( SELECT *, random() FROM ( WITH cte_1 AS (SELECT *, random() FROM citus_local_table), cte_2 AS (SELECT * FROM distributed_table) SELECT count(*) FROM cte_1, cte_2 ) as subquery_inner ) as subquery_top; count --------------------------------------------------------------------- 1 (1 row) SELECT clear_and_init_test_tables(); clear_and_init_test_tables --------------------------------------------------------------------- (1 row) -- prepared statement PREPARE citus_local_only AS SELECT count(*) FROM citus_local_table; -- execute 6 times, local tables without params EXECUTE citus_local_only; count --------------------------------------------------------------------- 6 (1 row) EXECUTE citus_local_only; count --------------------------------------------------------------------- 6 (1 row) EXECUTE citus_local_only; count --------------------------------------------------------------------- 6 (1 row) EXECUTE citus_local_only; count --------------------------------------------------------------------- 6 (1 row) EXECUTE citus_local_only; count --------------------------------------------------------------------- 6 (1 row) EXECUTE citus_local_only; count --------------------------------------------------------------------- 6 (1 row) -- execute 6 times, with param PREPARE citus_local_only_p(int) AS SELECT count(*) FROM citus_local_table WHERE a = $1; EXECUTE citus_local_only_p(1); count --------------------------------------------------------------------- 1 (1 row) EXECUTE citus_local_only_p(1); count --------------------------------------------------------------------- 1 (1 row) EXECUTE citus_local_only_p(1); count --------------------------------------------------------------------- 1 (1 row) EXECUTE citus_local_only_p(1); count --------------------------------------------------------------------- 1 (1 row) EXECUTE citus_local_only_p(1); count --------------------------------------------------------------------- 1 (1 row) EXECUTE citus_local_only_p(1); count --------------------------------------------------------------------- 1 (1 row) -- do not evalute the function -- show the logs EXECUTE citus_local_only_p(random()); count --------------------------------------------------------------------- 1 (1 row) EXECUTE citus_local_only_p(random()); count --------------------------------------------------------------------- 1 (1 row) PREPARE mixed_query(int, int, int) AS WITH cte_citus_local_table AS (SELECT * FROM citus_local_table WHERE a = $1), cte_postgres_local_table AS (SELECT * FROM postgres_local_table WHERE a = $2), cte_distributed_table AS (SELECT * FROM distributed_table WHERE a = $3), cte_mixes AS (SELECT * FROM cte_distributed_table, cte_citus_local_table, cte_postgres_local_table) SELECT count(*) FROM cte_mixes; EXECUTE mixed_query(1,2,3); count --------------------------------------------------------------------- 1 (1 row) EXECUTE mixed_query(1,2,3); count --------------------------------------------------------------------- 1 (1 row) EXECUTE mixed_query(1,2,3); count --------------------------------------------------------------------- 1 (1 row) EXECUTE mixed_query(1,2,3); count --------------------------------------------------------------------- 1 (1 row) EXECUTE mixed_query(1,2,3); count --------------------------------------------------------------------- 1 (1 row) EXECUTE mixed_query(1,2,3); count --------------------------------------------------------------------- 1 (1 row) EXECUTE mixed_query(1,2,3); count --------------------------------------------------------------------- 1 (1 row) SELECT clear_and_init_test_tables(); clear_and_init_test_tables --------------------------------------------------------------------- (1 row) -- anonymous columns WITH a AS (SELECT a, '' FROM citus_local_table GROUP BY a) SELECT a.a FROM a ORDER BY 1 LIMIT 5; a --------------------------------------------------------------------- 0 1 2 3 4 (5 rows) WITH a AS (SELECT b, '' FROM citus_local_table WHERE a = 1) SELECT * FROM a, a b ORDER BY 1 LIMIT 5; b | ?column? | b | ?column? --------------------------------------------------------------------- 1 | | 1 | (1 row) -- set operations should just work SELECT * FROM citus_local_table UNION SELECT * FROM postgres_local_table UNION SELECT * FROM distributed_table ORDER BY 1,2; a | b --------------------------------------------------------------------- 0 | 0 1 | 1 2 | 2 3 | 3 4 | 4 5 | 5 (6 rows) (SELECT * FROM citus_local_table ORDER BY 1,2 LIMIT 5) INTERSECT (SELECT i, i FROM generate_series(0, 100) i) ORDER BY 1, 2; a | b --------------------------------------------------------------------- 0 | 0 1 | 1 2 | 2 3 | 3 4 | 4 (5 rows) -- should just work as recursive planner kicks in SELECT count(*) FROM distributed_table WHERE a IN (SELECT a FROM citus_local_table); count --------------------------------------------------------------------- 6 (1 row) SELECT count(*) FROM citus_local_table WHERE a IN (SELECT a FROM distributed_table); count --------------------------------------------------------------------- 6 (1 row) SELECT count(*) FROM reference_table WHERE a IN (SELECT a FROM citus_local_table); count --------------------------------------------------------------------- 6 (1 row) SELECT count(*) FROM citus_local_table WHERE a IN (SELECT a FROM reference_table); count --------------------------------------------------------------------- 6 (1 row) -- nested recursive queries should just work SELECT count(*) FROM citus_local_table WHERE a IN (SELECT a FROM distributed_table WHERE a IN (SELECT b FROM citus_local_table WHERE b IN (SELECT b FROM postgres_local_table))); count --------------------------------------------------------------------- 6 (1 row) -- local outer joins SELECT count(*) FROM citus_local_table LEFT JOIN reference_table ON (true); count --------------------------------------------------------------------- 36 (1 row) SELECT count(*) FROM reference_table LEFT JOIN citus_local_table ON (true) LEFT JOIN postgres_local_table ON (true) LEFT JOIN reference_table r2 ON (true); count --------------------------------------------------------------------- 1296 (1 row) -- supported outer join SELECT count(*) FROM citus_local_table LEFT JOIN distributed_table ON (true); count --------------------------------------------------------------------- 36 (1 row) -- distinct in subquery on CTE WITH one_row AS ( SELECT a from citus_local_table WHERE b = 1 ) SELECT * FROM distributed_table WHERE b IN (SELECT DISTINCT a FROM one_row) ORDER BY 1, 2 LIMIT 1; a | b --------------------------------------------------------------------- 1 | 1 (1 row) WITH one_row_2 AS ( SELECT a from distributed_table WHERE b = 1 ) SELECT * FROM citus_local_table WHERE b IN (SELECT DISTINCT a FROM one_row_2) ORDER BY 1 ,2 LIMIT 1; a | b --------------------------------------------------------------------- 1 | 1 (1 row) -- join between citus local tables and distributed tables would fail SELECT count(*) FROM citus_local_table, distributed_table; count --------------------------------------------------------------------- 36 (1 row) SELECT * FROM citus_local_table, distributed_table ORDER BY 1,2,3,4 FOR UPDATE; ERROR: could not run distributed query with FOR UPDATE/SHARE commands -- join between citus local table and postgres local table would fail -- as citus local table is on the coordinator SELECT count(citus_local_table.b), count(postgres_local_table.a) FROM citus_local_table, postgres_local_table WHERE citus_local_table.a = postgres_local_table.b; ERROR: direct joins between distributed and local tables are not supported -- select for update is just OK SELECT * FROM citus_local_table ORDER BY 1,2 FOR UPDATE; a | b --------------------------------------------------------------------- 0 | 0 1 | 1 2 | 2 3 | 3 4 | 4 5 | 5 (6 rows) --------------------------------------------------------------------- ----- INSERT SELECT ----- --------------------------------------------------------------------- -- simple INSERT SELECT is OK SELECT clear_and_init_test_tables(); clear_and_init_test_tables --------------------------------------------------------------------- (1 row) INSERT INTO citus_local_table SELECT * from reference_table; INSERT INTO reference_table SELECT * from citus_local_table; INSERT INTO citus_local_table SELECT * from distributed_table; INSERT INTO distributed_table SELECT * from citus_local_table; INSERT INTO citus_local_table SELECT * from citus_local_table_2; INSERT INTO citus_local_table SELECT * from citus_local_table_2 ORDER BY 1,2 LIMIT 10; INSERT INTO citus_local_table SELECT * from postgres_local_table; INSERT INTO postgres_local_table SELECT * from citus_local_table; -- INSERT SELECT with local joins are OK SELECT clear_and_init_test_tables(); clear_and_init_test_tables --------------------------------------------------------------------- (1 row) INSERT INTO citus_local_table SELECT reference_table.* FROM reference_table JOIN citus_local_table ON (true); INSERT INTO reference_table SELECT reference_table.* FROM reference_table JOIN citus_local_table ON (true); INSERT INTO reference_table SELECT reference_table.* FROM reference_table, postgres_local_table JOIN citus_local_table ON (true); SELECT clear_and_init_test_tables(); clear_and_init_test_tables --------------------------------------------------------------------- (1 row) INSERT INTO distributed_table SELECT reference_table.* FROM reference_table JOIN citus_local_table ON (true); INSERT INTO distributed_table SELECT reference_table.* FROM reference_table, postgres_local_table JOIN citus_local_table ON (true); INSERT INTO postgres_local_table SELECT reference_table.* FROM reference_table JOIN citus_local_table ON (true); -- INSERT SELECT that joins reference and distributed tables is also OK SELECT clear_and_init_test_tables(); clear_and_init_test_tables --------------------------------------------------------------------- (1 row) INSERT INTO citus_local_table SELECT reference_table.* FROM reference_table JOIN distributed_table ON (true); INSERT INTO citus_local_table SELECT reference_table.* FROM reference_table, distributed_table; -- INSERT SELECT that joins citus local and distributed table directly will fail .. INSERT INTO citus_local_table SELECT distributed_table.* FROM distributed_table JOIN citus_local_table ON (true); -- .. but when wrapped into a CTE, join works fine INSERT INTO citus_local_table SELECT distributed_table.* FROM distributed_table JOIN (WITH cte AS (SELECT * FROM citus_local_table) SELECT * FROM cte) as foo ON (true); -- multi row insert is OK INSERT INTO citus_local_table VALUES (1, 2), (3, 4); --------------------------------------------------------------------- ----- DELETE / UPDATE ----- --------------------------------------------------------------------- -- modifications using citus local tables and postgres local tables -- are not supported, see below four tests SELECT clear_and_init_test_tables(); clear_and_init_test_tables --------------------------------------------------------------------- (1 row) DELETE FROM citus_local_table USING postgres_local_table WHERE citus_local_table.b = postgres_local_table.b; ERROR: Modifying local tables with remote local tables is not supported. UPDATE citus_local_table SET b = 5 FROM postgres_local_table WHERE citus_local_table.a = 3 AND citus_local_table.b = postgres_local_table.b; ERROR: Modifying local tables with remote local tables is not supported. DELETE FROM postgres_local_table USING citus_local_table WHERE citus_local_table.b = postgres_local_table.b; ERROR: Modifying local tables with remote local tables is not supported. UPDATE postgres_local_table SET b = 5 FROM citus_local_table WHERE citus_local_table.a = 3 AND citus_local_table.b = postgres_local_table.b; ERROR: Modifying local tables with remote local tables is not supported. -- no direct joins supported UPDATE distributed_table SET b = 6 FROM citus_local_table WHERE citus_local_table.a = distributed_table.a; UPDATE reference_table SET b = 6 FROM citus_local_table WHERE citus_local_table.a = reference_table.a; -- should not work, add HINT use CTEs UPDATE citus_local_table SET b = 6 FROM distributed_table WHERE citus_local_table.a = distributed_table.a; -- should work, add HINT use CTEs UPDATE citus_local_table SET b = 6 FROM reference_table WHERE citus_local_table.a = reference_table.a; -- should not work, add HINT use CTEs DELETE FROM distributed_table USING citus_local_table WHERE citus_local_table.a = distributed_table.a; -- should not work, add HINT use CTEs DELETE FROM citus_local_table USING distributed_table WHERE citus_local_table.a = distributed_table.a; DELETE FROM reference_table USING citus_local_table WHERE citus_local_table.a = reference_table.a; -- should work, add HINT use CTEs DELETE FROM citus_local_table USING reference_table WHERE citus_local_table.a = reference_table.a; -- just works DELETE FROM citus_local_table WHERE citus_local_table.a IN (SELECT a FROM distributed_table); -- just works DELETE FROM citus_local_table WHERE citus_local_table.a IN (SELECT a FROM reference_table); -- just works WITH distributed_table_cte AS (SELECT * FROM distributed_table) UPDATE citus_local_table SET b = 6 FROM distributed_table_cte WHERE citus_local_table.a = distributed_table_cte.a; -- just works WITH reference_table_cte AS (SELECT * FROM reference_table) UPDATE citus_local_table SET b = 6 FROM reference_table_cte WHERE citus_local_table.a = reference_table_cte.a; --------------------------------------------------------------------- ----- VIEW QUERIES ----- --------------------------------------------------------------------- CREATE MATERIALIZED VIEW mat_view_4 AS SELECT count(*) FROM citus_local_table JOIN reference_table USING (a); -- ok SELECT count(*) FROM mat_view_4; count --------------------------------------------------------------------- 1 (1 row) -- should work SELECT count(*) FROM distributed_table WHERE b in (SELECT count FROM mat_view_4); count --------------------------------------------------------------------- 0 (1 row) SET citus.enable_ddl_propagation TO OFF; CREATE VIEW view_2 AS SELECT count(*) FROM citus_local_table JOIN citus_local_table_2 USING (a) JOIN distributed_table USING (a); RESET citus.enable_ddl_propagation; -- should fail as view contains direct local dist join SELECT count(*) FROM view_2; count --------------------------------------------------------------------- 1 (1 row) SET citus.enable_ddl_propagation TO OFF; CREATE VIEW view_3 AS SELECT count(*) FROM citus_local_table_2 JOIN reference_table USING (a); RESET citus.enable_ddl_propagation; -- ok SELECT count(*) FROM view_3; count --------------------------------------------------------------------- 1 (1 row) -- view treated as subquery, so should work SELECT count(*) FROM view_3, distributed_table; count --------------------------------------------------------------------- 0 (1 row) --------------------------------------------------------------------- -- Some other tests with subqueries & CTE's -- --------------------------------------------------------------------- SELECT clear_and_init_test_tables(); clear_and_init_test_tables --------------------------------------------------------------------- (1 row) SELECT count(*) AS a, count(*) AS b FROM reference_table JOIN (SELECT count(*) as a, count(*) as b FROM citus_local_table_2 JOIN (SELECT count(*) as a, count(*) as b FROM postgres_local_table JOIN (SELECT count(*) as a, count(*) as b FROM reference_table as table_4677) subquery5108 USING (a)) subquery7132 USING (b)) subquery7294 USING (a); a | b --------------------------------------------------------------------- 1 | 1 (1 row) -- direct join inside CTE not supported WITH cte AS ( UPDATE citus_local_table lt SET a = mt.a FROM distributed_table mt WHERE mt.b = lt.b RETURNING lt.b, lt.a ) SELECT * FROM cte JOIN distributed_table mt ON mt.b = cte.b ORDER BY 1,2,3,4; b | a | a | b --------------------------------------------------------------------- 0 | 0 | 0 | 0 1 | 1 | 1 | 1 2 | 2 | 2 | 2 3 | 3 | 3 | 3 4 | 4 | 4 | 4 5 | 5 | 5 | 5 (6 rows) -- join with CTE just works UPDATE citus_local_table SET a=5 FROM (SELECT avg(distributed_table.b) as avg_b FROM distributed_table) as foo WHERE foo.avg_b = citus_local_table.b; -- should work UPDATE distributed_table SET b = avg_a FROM (SELECT avg(citus_local_table.a) as avg_a FROM citus_local_table) as foo WHERE foo.avg_a = distributed_table.a RETURNING distributed_table.*; a | b --------------------------------------------------------------------- (0 rows) -- it is unfortunate that recursive planner cannot detect this -- but expected to not work UPDATE citus_local_table SET a=5 FROM (SELECT b FROM distributed_table) AS foo WHERE foo.b = citus_local_table.b; ERROR: local table citus_local_table cannot be joined with these distributed tables --------------------------------------------------------------------- -- test different execution paths -- --------------------------------------------------------------------- -- a bit different explain output than for postgres local tables EXPLAIN (COSTS FALSE) INSERT INTO citus_local_table SELECT * FROM distributed_table ORDER BY distributed_table.* LIMIT 10; QUERY PLAN --------------------------------------------------------------------- Custom Scan (Citus INSERT ... SELECT) INSERT/SELECT method: pull to coordinator -> Limit -> Sort Sort Key: remote_scan.worker_column_3 -> Custom Scan (Citus Adaptive) Task Count: 4 Tasks Shown: One of 4 -> Task Node: host=localhost port=xxxxx dbname=regression -> Limit -> Sort Sort Key: distributed_table.* -> Seq Scan on distributed_table_1510004 distributed_table (14 rows) -- show that we do not pull to coordinator EXPLAIN (COSTS FALSE) INSERT INTO citus_local_table SELECT * FROM citus_local_table; QUERY PLAN --------------------------------------------------------------------- Custom Scan (Citus Adaptive) Task Count: 1 Tasks Shown: All -> Task Node: host=localhost port=xxxxx dbname=regression -> Insert on citus_local_table_1510001 citus_table_alias -> Seq Scan on citus_local_table_1510001 citus_local_table (7 rows) EXPLAIN (COSTS FALSE) INSERT INTO citus_local_table SELECT reference_table.* FROM reference_table; QUERY PLAN --------------------------------------------------------------------- Custom Scan (Citus Adaptive) Task Count: 1 Tasks Shown: All -> Task Node: host=localhost port=xxxxx dbname=regression -> Insert on citus_local_table_1510001 citus_table_alias -> Seq Scan on reference_table_1510003 reference_table (7 rows) EXPLAIN (COSTS FALSE) INSERT INTO citus_local_table SELECT reference_table.* FROM reference_table, postgres_local_table; QUERY PLAN --------------------------------------------------------------------- Custom Scan (Citus INSERT ... SELECT) INSERT/SELECT method: pull to coordinator -> Custom Scan (Citus Adaptive) Task Count: 1 Tasks Shown: All -> Task Node: host=localhost port=xxxxx dbname=regression -> Nested Loop -> Seq Scan on reference_table_1510003 reference_table -> Materialize -> Seq Scan on postgres_local_table (11 rows) -- show that we pull to coordinator when a distributed table is involved EXPLAIN (COSTS FALSE) INSERT INTO citus_local_table SELECT reference_table.* FROM reference_table, distributed_table; QUERY PLAN --------------------------------------------------------------------- Custom Scan (Citus INSERT ... SELECT) INSERT/SELECT method: pull to coordinator -> Custom Scan (Citus Adaptive) Task Count: 4 Tasks Shown: One of 4 -> Task Node: host=localhost port=xxxxx dbname=regression -> Nested Loop -> Seq Scan on distributed_table_1510004 distributed_table -> Materialize -> Seq Scan on reference_table_1510003 reference_table (11 rows) -- truncate tables & add unique constraints to be able to define foreign keys TRUNCATE reference_table, citus_local_table, distributed_table; \c - - - :master_port SET search_path TO citus_local_table_queries_mx; SET citus.shard_replication_factor TO 1; ALTER TABLE reference_table ADD CONSTRAINT pkey_ref PRIMARY KEY (a); ALTER TABLE citus_local_table ADD CONSTRAINT pkey_c PRIMARY KEY (a); -- define a foreign key chain distributed table -> reference table -> citus local table -- to test sequential execution ALTER TABLE distributed_table ADD CONSTRAINT fkey_dist_to_ref FOREIGN KEY(a) REFERENCES reference_table(a) ON DELETE RESTRICT; ALTER TABLE reference_table ADD CONSTRAINT fkey_ref_to_local FOREIGN KEY(a) REFERENCES citus_local_table(a) ON DELETE RESTRICT; \c - - - :worker_1_port SET search_path TO citus_local_table_queries_mx; INSERT INTO citus_local_table VALUES (1); INSERT INTO reference_table VALUES (1); BEGIN; INSERT INTO citus_local_table VALUES (1) ON CONFLICT (a) DO NOTHING; INSERT INTO distributed_table VALUES (1); -- should show sequential as first inserting into citus local table -- would force the xact block to use sequential execution show citus.multi_shard_modify_mode; citus.multi_shard_modify_mode --------------------------------------------------------------------- sequential (1 row) ROLLBACK; BEGIN; TRUNCATE distributed_table; -- should error out as we truncated distributed_table via parallel execution TRUNCATE citus_local_table CASCADE; NOTICE: truncate cascades to table "reference_table" NOTICE: truncate cascades to table "distributed_table" ERROR: cannot execute DDL on table "citus_local_table" because there was a parallel DDL access to distributed table "distributed_table" in the same transaction ROLLBACK; BEGIN; SET LOCAL citus.multi_shard_modify_mode TO 'sequential'; TRUNCATE distributed_table; -- should work fine as we already switched to sequential execution -- before parallel truncate TRUNCATE citus_local_table CASCADE; NOTICE: truncate cascades to table "reference_table" NOTICE: truncate cascades to table "distributed_table" NOTICE: truncate cascades to table "distributed_table_xxxxx" NOTICE: truncate cascades to table "distributed_table_xxxxx" ROLLBACK; \c - - - :master_port SET search_path TO citus_local_table_queries_mx; SET citus.shard_replication_factor TO 1; ALTER TABLE distributed_table DROP CONSTRAINT fkey_dist_to_ref; \c - - - :worker_1_port SET search_path TO citus_local_table_queries_mx; BEGIN; INSERT INTO citus_local_table VALUES (1) ON CONFLICT (a) DO NOTHING; show citus.multi_shard_modify_mode; citus.multi_shard_modify_mode --------------------------------------------------------------------- sequential (1 row) ROLLBACK; \c - - - :master_port SET search_path TO citus_local_table_queries_mx; SET citus.shard_replication_factor TO 1; -- remove uniqueness constraint and dependent foreign key constraint for next tests ALTER TABLE reference_table DROP CONSTRAINT fkey_ref_to_local; ALTER TABLE citus_local_table DROP CONSTRAINT pkey_c; \c - - - :worker_1_port SET search_path TO citus_local_table_queries_mx; COPY citus_local_table(a) FROM PROGRAM 'seq 1'; BEGIN; COPY citus_local_table(a) FROM PROGRAM 'seq 1'; COPY citus_local_table(a) FROM PROGRAM 'seq 1'; COMMIT; COPY citus_local_table TO STDOUT; 1 \N 1 \N 1 \N 1 \N COPY (SELECT * FROM citus_local_table) TO STDOUT; 1 \N 1 \N 1 \N 1 \N BEGIN; COPY citus_local_table TO STDOUT; 1 \N 1 \N 1 \N 1 \N COMMIT; BEGIN; COPY (SELECT * FROM citus_local_table) TO STDOUT; 1 \N 1 \N 1 \N 1 \N COMMIT; -- truncate test tables for next test TRUNCATE citus_local_table, reference_table, distributed_table; BEGIN; INSERT INTO citus_local_table VALUES (1), (2); SAVEPOINT sp1; INSERT INTO citus_local_table VALUES (3), (4); ROLLBACK TO SAVEPOINT sp1; SELECT * FROM citus_local_table ORDER BY 1,2; a | b --------------------------------------------------------------------- 1 | 2 | (2 rows) SAVEPOINT sp2; INSERT INTO citus_local_table VALUES (3), (4); INSERT INTO distributed_table VALUES (3), (4); ROLLBACK TO SAVEPOINT sp2; SELECT * FROM citus_local_table ORDER BY 1,2; a | b --------------------------------------------------------------------- 1 | 2 | (2 rows) SELECT * FROM distributed_table ORDER BY 1,2; a | b --------------------------------------------------------------------- (0 rows) SAVEPOINT sp3; INSERT INTO citus_local_table VALUES (3), (2); INSERT INTO reference_table VALUES (3), (2); ROLLBACK TO SAVEPOINT sp3; SELECT * FROM citus_local_table ORDER BY 1,2; a | b --------------------------------------------------------------------- 1 | 2 | (2 rows) SELECT * FROM reference_table ORDER BY 1,2; a | b --------------------------------------------------------------------- (0 rows) COMMIT; \c - - - :master_port -- cleanup at exit DROP SCHEMA citus_local_table_queries_mx CASCADE; NOTICE: drop cascades to 9 other objects