regress/expected/replicate_reference_tables_to_coordinator.out

--
-- REPLICATE_REF_TABLES_ON_COORDINATOR
--
CREATE SCHEMA replicate_ref_to_coordinator;
SET search_path TO 'replicate_ref_to_coordinator';
SET citus.shard_replication_factor TO 1;
SET citus.shard_count TO 4;
SET citus.next_shard_id TO 8000000;
SET citus.next_placement_id TO 8000000;
--- enable logging to see which tasks are executed locally
SET citus.log_local_commands TO ON;
CREATE TABLE squares(a int, b int);
SELECT create_reference_table('squares');
 create_reference_table
---------------------------------------------------------------------

(1 row)

INSERT INTO squares SELECT i, i * i FROM generate_series(1, 10) i;
NOTICE:  executing the copy locally for shard xxxxx
CREATE INDEX CONCURRENTLY squares_a_idx ON squares (a);
SELECT substring(current_Setting('server_version'), '\d+')::int > 11 AS server_version_above_eleven
\gset
\if :server_version_above_eleven
REINDEX INDEX CONCURRENTLY squares_a_idx;
\endif
DROP INDEX CONCURRENTLY squares_a_idx;
-- should be executed locally
SELECT count(*) FROM squares;
NOTICE:  executing the command locally: SELECT count(*) AS count FROM replicate_ref_to_coordinator.squares_8000000 squares
 count
---------------------------------------------------------------------
    10
(1 row)

-- create a second reference table
CREATE TABLE numbers(a int);
SELECT create_reference_table('numbers');
 create_reference_table
---------------------------------------------------------------------

(1 row)

INSERT INTO numbers VALUES (20), (21);
NOTICE:  executing the command locally: INSERT INTO replicate_ref_to_coordinator.numbers_8000001 AS citus_table_alias (a) VALUES (20), (21)
CREATE OR REPLACE FUNCTION my_volatile_fn()
RETURNS INT AS $$
BEGIN
  RETURN 1;
END; $$ language plpgsql VOLATILE;
-- INSERT ... SELECT between reference tables
BEGIN;
EXPLAIN (COSTS OFF) INSERT INTO squares SELECT a, a*a FROM numbers;
                        QUERY PLAN
---------------------------------------------------------------------
 Custom Scan (Citus Adaptive)
   Task Count: 1
   Tasks Shown: All
   ->  Task
         Node: host=localhost port=xxxxx dbname=regression
         ->  Insert on squares_8000000 citus_table_alias
               ->  Seq Scan on numbers_8000001 numbers
(7 rows)

INSERT INTO squares SELECT a, a*a FROM numbers;
SELECT * FROM squares WHERE a >= 20 ORDER BY a;
 a  |  b
---------------------------------------------------------------------
 20 | 400
 21 | 441
(2 rows)

ROLLBACK;
BEGIN;
EXPLAIN (COSTS OFF) INSERT INTO numbers SELECT a FROM squares WHERE a < 3;
                        QUERY PLAN
---------------------------------------------------------------------
 Custom Scan (Citus Adaptive)
   Task Count: 1
   Tasks Shown: All
   ->  Task
         Node: host=localhost port=xxxxx dbname=regression
         ->  Insert on numbers_8000001 citus_table_alias
               ->  Seq Scan on squares_8000000 squares
                     Filter: (a < 3)
(8 rows)

INSERT INTO numbers SELECT a FROM squares WHERE a < 3;
SELECT * FROM numbers ORDER BY a;
 a
---------------------------------------------------------------------
  1
  2
 20
 21
(4 rows)

ROLLBACK;
-- Make sure we hide shard tables ...
SELECT citus_table_is_visible('numbers_8000001'::regclass::oid);
 citus_table_is_visible
---------------------------------------------------------------------
 f
(1 row)

-- Join between reference tables and local tables
CREATE TABLE local_table(a int);
INSERT INTO local_table VALUES (2), (4), (7), (20);
EXPLAIN (COSTS OFF) SELECT local_table.a, numbers.a FROM local_table NATURAL JOIN numbers;
                         QUERY PLAN
---------------------------------------------------------------------
 Custom Scan (Citus Adaptive)
   Task Count: 1
   Tasks Shown: All
   ->  Task
         Node: host=localhost port=xxxxx dbname=regression
         ->  Merge Join
               Merge Cond: (local_table.a = numbers.a)
               ->  Sort
                     Sort Key: local_table.a
                     ->  Seq Scan on local_table
               ->  Sort
                     Sort Key: numbers.a
                     ->  Seq Scan on numbers_8000001 numbers
(13 rows)

SELECT local_table.a, numbers.a FROM local_table NATURAL JOIN numbers ORDER BY 1;
NOTICE:  executing the command locally: SELECT local_table.a, numbers.a FROM (replicate_ref_to_coordinator.local_table JOIN replicate_ref_to_coordinator.numbers_8000001 numbers(a) USING (a)) ORDER BY local_table.a
 a  | a
---------------------------------------------------------------------
 20 | 20
(1 row)

-- test non equijoin
SELECT lt.a, sq.a, sq.b
FROM local_table lt
JOIN squares sq ON sq.a > lt.a and sq.b > 90
ORDER BY 1,2,3;
NOTICE:  executing the command locally: SELECT lt.a, sq.a, sq.b FROM (replicate_ref_to_coordinator.local_table lt JOIN replicate_ref_to_coordinator.squares_8000000 sq ON (((sq.a OPERATOR(pg_catalog.>) lt.a) AND (sq.b OPERATOR(pg_catalog.>) 90)))) ORDER BY lt.a, sq.a, sq.b
 a | a  |  b
---------------------------------------------------------------------
 2 | 10 | 100
 4 | 10 | 100
 7 | 10 | 100
(3 rows)

-- should work if in transaction block
BEGIN;
SELECT local_table.a, numbers.a FROM local_table NATURAL JOIN numbers ORDER BY 1;
NOTICE:  executing the command locally: SELECT local_table.a, numbers.a FROM (replicate_ref_to_coordinator.local_table JOIN replicate_ref_to_coordinator.numbers_8000001 numbers(a) USING (a)) ORDER BY local_table.a
 a  | a
---------------------------------------------------------------------
 20 | 20
(1 row)

ROLLBACK;
-- should work if in a DO block
DO $$
BEGIN
	PERFORM local_table.a, numbers.a FROM local_table NATURAL JOIN numbers;
END;
$$;
NOTICE:  executing the command locally: SELECT local_table.a, numbers.a FROM (replicate_ref_to_coordinator.local_table JOIN replicate_ref_to_coordinator.numbers_8000001 numbers(a) USING (a))
CONTEXT:  SQL statement "SELECT local_table.a, numbers.a FROM local_table NATURAL JOIN numbers"
PL/pgSQL function inline_code_block line XX at PERFORM
-- test plpgsql function
CREATE FUNCTION test_reference_local_join_plpgsql_func()
RETURNS void AS $$
BEGIN
	INSERT INTO local_table VALUES (21);
	INSERT INTO numbers VALUES (4);
	PERFORM local_table.a, numbers.a FROM local_table NATURAL JOIN numbers ORDER BY 1;
	RAISE EXCEPTION '';
	PERFORM local_table.a, numbers.a FROM local_table NATURAL JOIN numbers ORDER BY 1;
END;
$$ LANGUAGE plpgsql;
SELECT test_reference_local_join_plpgsql_func();
NOTICE:  executing the command locally: INSERT INTO replicate_ref_to_coordinator.numbers_8000001 (a) VALUES (4)
CONTEXT:  SQL statement "INSERT INTO numbers VALUES (4)"
PL/pgSQL function test_reference_local_join_plpgsql_func() line XX at SQL statement
NOTICE:  executing the command locally: SELECT local_table.a, numbers.a FROM (replicate_ref_to_coordinator.local_table JOIN replicate_ref_to_coordinator.numbers_8000001 numbers(a) USING (a)) ORDER BY local_table.a
CONTEXT:  SQL statement "SELECT local_table.a, numbers.a FROM local_table NATURAL JOIN numbers ORDER BY 1"
PL/pgSQL function test_reference_local_join_plpgsql_func() line XX at PERFORM
ERROR:
CONTEXT:  PL/pgSQL function test_reference_local_join_plpgsql_func() line XX at RAISE
SELECT sum(a) FROM local_table;
 sum
---------------------------------------------------------------------
  33
(1 row)

SELECT sum(a) FROM numbers;
NOTICE:  executing the command locally: SELECT sum(a) AS sum FROM replicate_ref_to_coordinator.numbers_8000001 numbers
 sum
---------------------------------------------------------------------
  41
(1 row)

-- error if in procedure's subtransaction
CREATE PROCEDURE test_reference_local_join_proc() AS $$
SELECT local_table.a, numbers.a FROM local_table NATURAL JOIN numbers ORDER BY 1;
$$ LANGUAGE sql;
CALL test_reference_local_join_proc();
NOTICE:  executing the command locally: SELECT local_table.a, numbers.a FROM (replicate_ref_to_coordinator.local_table JOIN replicate_ref_to_coordinator.numbers_8000001 numbers(a) USING (a)) ORDER BY local_table.a
CONTEXT:  SQL function "test_reference_local_join_proc" statement 1
CREATE SCHEMA s1;
CREATE TABLE s1.ref(a int);
SELECT create_reference_table('s1.ref');
 create_reference_table
---------------------------------------------------------------------

(1 row)

BEGIN;
SELECT local_table.a, r.a FROM local_table NATURAL JOIN s1.ref r ORDER BY 1;
NOTICE:  executing the command locally: SELECT local_table.a, r.a FROM (replicate_ref_to_coordinator.local_table JOIN s1.ref_8000002 r(a) USING (a)) ORDER BY local_table.a
 a | a
---------------------------------------------------------------------
(0 rows)

ROLLBACK;
BEGIN;
WITH t1 AS (
	SELECT my_volatile_fn() r, a FROM local_table
) SELECT count(*) FROM t1, numbers WHERE t1.a = numbers.a AND r < 0.5;
NOTICE:  executing the command locally: WITH t1 AS (SELECT replicate_ref_to_coordinator.my_volatile_fn() AS r, local_table.a FROM replicate_ref_to_coordinator.local_table) SELECT count(*) AS count FROM t1, replicate_ref_to_coordinator.numbers_8000001 numbers WHERE ((t1.a OPERATOR(pg_catalog.=) numbers.a) AND ((t1.r)::numeric OPERATOR(pg_catalog.<) 0.5))
 count
---------------------------------------------------------------------
     0
(1 row)

END;
BEGIN;
WITH t1 AS (
	SELECT my_volatile_fn() r, a FROM numbers
) SELECT count(*) FROM t1, local_table WHERE t1.a = local_table.a AND r < 0.5;
NOTICE:  executing the command locally: WITH t1 AS (SELECT replicate_ref_to_coordinator.my_volatile_fn() AS r, numbers.a FROM replicate_ref_to_coordinator.numbers_8000001 numbers) SELECT count(*) AS count FROM t1, replicate_ref_to_coordinator.local_table WHERE ((t1.a OPERATOR(pg_catalog.=) local_table.a) AND ((t1.r)::numeric OPERATOR(pg_catalog.<) 0.5))
 count
---------------------------------------------------------------------
     0
(1 row)

END;
BEGIN;
SELECT count(*) FROM local_table
WHERE EXISTS(SELECT my_volatile_fn() FROM numbers WHERE local_table.a = numbers.a);
NOTICE:  executing the command locally: SELECT count(*) AS count FROM replicate_ref_to_coordinator.local_table WHERE (EXISTS (SELECT replicate_ref_to_coordinator.my_volatile_fn() AS my_volatile_fn FROM replicate_ref_to_coordinator.numbers_8000001 numbers WHERE (local_table.a OPERATOR(pg_catalog.=) numbers.a)))
 count
---------------------------------------------------------------------
     1
(1 row)

END;
BEGIN;
SELECT count(*) FROM numbers
WHERE EXISTS(SELECT my_volatile_fn() FROM local_table WHERE local_table.a = numbers.a);
NOTICE:  executing the command locally: SELECT count(*) AS count FROM replicate_ref_to_coordinator.numbers_8000001 numbers WHERE (EXISTS (SELECT replicate_ref_to_coordinator.my_volatile_fn() AS my_volatile_fn FROM replicate_ref_to_coordinator.local_table WHERE (local_table.a OPERATOR(pg_catalog.=) numbers.a)))
 count
---------------------------------------------------------------------
     1
(1 row)

END;
DROP SCHEMA s1 CASCADE;
NOTICE:  drop cascades to 2 other objects
DETAIL:  drop cascades to table s1.ref
drop cascades to table s1.ref_8000002
-- not error if inside a SQL UDF call
CREATE or replace FUNCTION test_reference_local_join_func()
RETURNS SETOF RECORD AS $$
SET LOCAL citus.enable_local_execution to false;
INSERT INTO numbers VALUES (2);
SELECT local_table.a, numbers.a FROM local_table NATURAL JOIN numbers ORDER BY 1;
$$ LANGUAGE sql;
SELECT test_reference_local_join_func();
 test_reference_local_join_func
---------------------------------------------------------------------
 (2,2)
 (20,20)
(2 rows)

-- CTEs are allowed
WITH ins AS (INSERT INTO numbers VALUES (1) RETURNING *)
SELECT * FROM numbers, local_table ORDER BY 1,2;
NOTICE:  executing the command locally: INSERT INTO replicate_ref_to_coordinator.numbers_8000001 (a) VALUES (1) RETURNING a
NOTICE:  executing the command locally: SELECT numbers.a, local_table.a FROM replicate_ref_to_coordinator.numbers_8000001 numbers, (SELECT local_table_1.a FROM (SELECT intermediate_result.a FROM read_intermediate_result('XXX_2'::text, 'binary'::citus_copy_format) intermediate_result(a integer)) local_table_1) local_table ORDER BY numbers.a, local_table.a
 a  | a
---------------------------------------------------------------------
  2 |  2
  2 |  4
  2 |  7
  2 | 20
 20 |  2
 20 |  4
 20 |  7
 20 | 20
 21 |  2
 21 |  4
 21 |  7
 21 | 20
(12 rows)

WITH t AS (SELECT *, my_volatile_fn() x FROM numbers FOR UPDATE)
SELECT * FROM numbers, local_table
WHERE EXISTS (SELECT * FROM t WHERE t.x = numbers.a);
NOTICE:  executing the command locally: WITH t AS (SELECT numbers_1.a, replicate_ref_to_coordinator.my_volatile_fn() AS x FROM replicate_ref_to_coordinator.numbers_8000001 numbers_1 FOR UPDATE OF numbers_1) SELECT numbers.a, local_table.a FROM replicate_ref_to_coordinator.numbers_8000001 numbers, replicate_ref_to_coordinator.local_table WHERE (EXISTS (SELECT t.a, t.x FROM t WHERE (t.x OPERATOR(pg_catalog.=) numbers.a)))
 a | a
---------------------------------------------------------------------
 1 |  2
 1 |  4
 1 |  7
 1 | 20
(4 rows)

WITH t AS (SELECT *, my_volatile_fn() x FROM numbers)
SELECT * FROM numbers, local_table
WHERE EXISTS (SELECT * FROM t WHERE t.x = numbers.a);
NOTICE:  executing the command locally: WITH t AS (SELECT numbers_1.a, replicate_ref_to_coordinator.my_volatile_fn() AS x FROM replicate_ref_to_coordinator.numbers_8000001 numbers_1) SELECT numbers.a, local_table.a FROM replicate_ref_to_coordinator.numbers_8000001 numbers, replicate_ref_to_coordinator.local_table WHERE (EXISTS (SELECT t.a, t.x FROM t WHERE (t.x OPERATOR(pg_catalog.=) numbers.a)))
 a | a
---------------------------------------------------------------------
 1 |  2
 1 |  4
 1 |  7
 1 | 20
(4 rows)

-- shouldn't plan locally even if distributed table is in CTE or subquery
CREATE TABLE dist(a int);
SELECT create_distributed_table('dist', 'a');
 create_distributed_table
---------------------------------------------------------------------

(1 row)

INSERT INTO dist VALUES (20),(30);
WITH t AS (SELECT *, my_volatile_fn() x FROM dist)
SELECT * FROM numbers, local_table
WHERE EXISTS (SELECT * FROM t WHERE t.x = numbers.a);
ERROR:  function replicate_ref_to_coordinator.my_volatile_fn() does not exist
HINT:  No function matches the given name and argument types. You might need to add explicit type casts.
CONTEXT:  while executing command on localhost:xxxxx
-- test CTE being reference/local join for distributed query
WITH t as (SELECT n.a, my_volatile_fn() x FROM numbers n NATURAL JOIN local_table l)
SELECT a FROM t NATURAL JOIN dist;
NOTICE:  executing the command locally: SELECT n.a, replicate_ref_to_coordinator.my_volatile_fn() AS x FROM (replicate_ref_to_coordinator.numbers_8000001 n(a) JOIN replicate_ref_to_coordinator.local_table l USING (a))
 a
---------------------------------------------------------------------
 20
(1 row)

 -- shouldn't error if FOR UPDATE/FOR SHARE
SELECT local_table.a, numbers.a FROM local_table NATURAL JOIN numbers FOR SHARE;
NOTICE:  executing the command locally: SELECT local_table.a, numbers.a FROM (replicate_ref_to_coordinator.local_table JOIN replicate_ref_to_coordinator.numbers_8000001 numbers(a) USING (a)) FOR SHARE OF local_table FOR SHARE OF numbers
 a  | a
---------------------------------------------------------------------
  2 |  2
 20 | 20
(2 rows)

SELECT local_table.a, numbers.a FROM local_table NATURAL JOIN numbers FOR UPDATE;
NOTICE:  executing the command locally: SELECT local_table.a, numbers.a FROM (replicate_ref_to_coordinator.local_table JOIN replicate_ref_to_coordinator.numbers_8000001 numbers(a) USING (a)) FOR UPDATE OF local_table FOR UPDATE OF numbers
 a  | a
---------------------------------------------------------------------
  2 |  2
 20 | 20
(2 rows)

--
-- Joins between reference tables and views shouldn't be planned locally.
--
CREATE VIEW numbers_v AS SELECT * FROM numbers WHERE a=1;
SELECT public.coordinator_plan($Q$
EXPLAIN (COSTS FALSE)
	SELECT * FROM squares JOIN numbers_v ON squares.a = numbers_v.a;
$Q$);
       coordinator_plan
---------------------------------------------------------------------
 Custom Scan (Citus Adaptive)
   Task Count: 1
(2 rows)

CREATE VIEW local_table_v AS SELECT * FROM local_table WHERE a BETWEEN 1 AND 10;
SELECT public.coordinator_plan($Q$
EXPLAIN (COSTS FALSE)
	SELECT * FROM squares JOIN local_table_v ON squares.a = local_table_v.a;
$Q$);
       coordinator_plan
---------------------------------------------------------------------
 Custom Scan (Citus Adaptive)
   Task Count: 1
(2 rows)

DROP VIEW numbers_v, local_table_v;
--
-- Joins between reference tables and materialized views are allowed to
-- be planned to be executed locally.
--
CREATE MATERIALIZED VIEW numbers_v AS SELECT * FROM numbers WHERE a BETWEEN 1 AND 10;
NOTICE:  executing the command locally: SELECT a FROM replicate_ref_to_coordinator.numbers_8000001 numbers WHERE ((a OPERATOR(pg_catalog.>=) 1) AND (a OPERATOR(pg_catalog.<=) 10))
REFRESH MATERIALIZED VIEW numbers_v;
NOTICE:  executing the command locally: SELECT numbers.a FROM replicate_ref_to_coordinator.numbers_8000001 numbers WHERE ((numbers.a OPERATOR(pg_catalog.>=) 1) AND (numbers.a OPERATOR(pg_catalog.<=) 10))
SELECT * FROM squares JOIN numbers_v ON squares.a = numbers_v.a ORDER BY 1;
NOTICE:  executing the command locally: SELECT squares.a, squares.b, numbers_v.a FROM (replicate_ref_to_coordinator.squares_8000000 squares JOIN replicate_ref_to_coordinator.numbers_v ON ((squares.a OPERATOR(pg_catalog.=) numbers_v.a))) ORDER BY squares.a
 a | b | a
---------------------------------------------------------------------
 1 | 1 | 1
 2 | 4 | 2
(2 rows)

--
-- Joins between reference tables, local tables, and function calls
-- are allowed
--
SELECT count(*)
FROM local_table a, numbers b, generate_series(1, 10) c
WHERE a.a = b.a AND a.a = c;
NOTICE:  executing the command locally: SELECT count(*) AS count FROM replicate_ref_to_coordinator.local_table a, replicate_ref_to_coordinator.numbers_8000001 b, generate_series(1, 10) c(c) WHERE ((a.a OPERATOR(pg_catalog.=) b.a) AND (a.a OPERATOR(pg_catalog.=) c.c))
 count
---------------------------------------------------------------------
     1
(1 row)

-- and it should be okay if the function call is not a data source
SELECT abs(a.a) FROM local_table a, numbers b WHERE a.a = b.a;
NOTICE:  executing the command locally: SELECT abs(a.a) AS abs FROM replicate_ref_to_coordinator.local_table a, replicate_ref_to_coordinator.numbers_8000001 b WHERE (a.a OPERATOR(pg_catalog.=) b.a)
 abs
---------------------------------------------------------------------
   2
  20
(2 rows)

SELECT a.a FROM local_table a, numbers b WHERE a.a = b.a ORDER BY abs(a.a);
NOTICE:  executing the command locally: SELECT a.a FROM replicate_ref_to_coordinator.local_table a, replicate_ref_to_coordinator.numbers_8000001 b WHERE (a.a OPERATOR(pg_catalog.=) b.a) ORDER BY (abs(a.a))
 a
---------------------------------------------------------------------
  2
 20
(2 rows)

TRUNCATE local_table;
TRUNCATE numbers;
NOTICE:  executing the command locally: TRUNCATE TABLE replicate_ref_to_coordinator.numbers_xxxxx CASCADE
BEGIN;
INSERT INTO local_table VALUES (1), (2), (3), (4);
INSERT INTO numbers VALUES (1), (2), (3), (4);
NOTICE:  executing the command locally: INSERT INTO replicate_ref_to_coordinator.numbers_8000001 AS citus_table_alias (a) VALUES (1), (2), (3), (4)
ALTER TABLE numbers ADD COLUMN d int;
NOTICE:  executing the command locally: SELECT worker_apply_shard_ddl_command (8000001, 'replicate_ref_to_coordinator', 'ALTER TABLE numbers ADD COLUMN d int;')
SELECT * FROM local_table JOIN numbers USING(a) ORDER BY a;
NOTICE:  executing the command locally: SELECT local_table.a, numbers.d FROM (replicate_ref_to_coordinator.local_table JOIN replicate_ref_to_coordinator.numbers_8000001 numbers(a, d) USING (a)) ORDER BY local_table.a
 a | d
---------------------------------------------------------------------
 1 |
 2 |
 3 |
 4 |
(4 rows)

ROLLBACK;
BEGIN;
INSERT INTO local_table VALUES (1), (2), (3);
WITH t as (SELECT n.a, my_volatile_fn() x FROM numbers n NATURAL JOIN local_table l ORDER BY n.a, x)
SELECT a FROM t NATURAL JOIN dist ORDER BY a;
NOTICE:  executing the command locally: SELECT n.a, replicate_ref_to_coordinator.my_volatile_fn() AS x FROM (replicate_ref_to_coordinator.numbers_8000001 n(a) JOIN replicate_ref_to_coordinator.local_table l USING (a)) ORDER BY n.a, (replicate_ref_to_coordinator.my_volatile_fn())
 a
---------------------------------------------------------------------
(0 rows)

ROLLBACK;
BEGIN;
INSERT INTO local_table VALUES (1), (2), (3);
INSERT INTO numbers SELECT * FROM generate_series(1, 100);
NOTICE:  executing the copy locally for shard xxxxx
INSERT INTO numbers SELECT * FROM numbers;
NOTICE:  executing the command locally: INSERT INTO replicate_ref_to_coordinator.numbers_8000001 AS citus_table_alias (a) SELECT a FROM replicate_ref_to_coordinator.numbers_8000001 numbers
SELECT COUNT(*) FROM local_table JOIN numbers using (a);
NOTICE:  executing the command locally: SELECT count(*) AS count FROM (replicate_ref_to_coordinator.local_table JOIN replicate_ref_to_coordinator.numbers_8000001 numbers(a) USING (a))
 count
---------------------------------------------------------------------
     6
(1 row)

UPDATE numbers SET a = a + 1;
NOTICE:  executing the command locally: UPDATE replicate_ref_to_coordinator.numbers_8000001 numbers SET a = (a OPERATOR(pg_catalog.+) 1)
SELECT COUNT(*) FROM local_table JOIN numbers using (a);
NOTICE:  executing the command locally: SELECT count(*) AS count FROM (replicate_ref_to_coordinator.local_table JOIN replicate_ref_to_coordinator.numbers_8000001 numbers(a) USING (a))
 count
---------------------------------------------------------------------
     4
(1 row)

ROLLBACK;
-- verify that we can drop columns from reference tables replicated to the coordinator
-- see https://github.com/citusdata/citus/issues/3279
ALTER TABLE squares DROP COLUMN b;
NOTICE:  executing the command locally: SELECT worker_apply_shard_ddl_command (8000000, 'replicate_ref_to_coordinator', 'ALTER TABLE squares DROP COLUMN b;')
-- verify that we replicate the reference tables that are distributed before
-- adding the coordinator as a worker.
SELECT master_remove_node('localhost', :master_port);
 master_remove_node
---------------------------------------------------------------------

(1 row)

-- add the coordinator as a worker node and verify that the reference tables are replicated
SELECT master_add_node('localhost', :master_port, groupid => 0) AS master_nodeid \gset
NOTICE:  Replicating reference table "squares" to the node localhost:xxxxx
NOTICE:  Replicating reference table "numbers" to the node localhost:xxxxx
-- clean-up
SET client_min_messages TO ERROR;
DROP SCHEMA replicate_ref_to_coordinator CASCADE;
-- Make sure the shard was dropped
SELECT 'numbers_8000001'::regclass::oid;
ERROR:  relation "numbers_8000001" does not exist
SET search_path TO DEFAULT;
RESET client_min_messages;