Ensure schema exists on reference table creation

If the schema doesn't exists on the workers, create it.

src/backend/distributed/commands/create_distributed_table.c

@@ -236,6 +236,14 @@ create_reference_table(PG_FUNCTION_ARGS)
 	EnsureCoordinator();
 	CheckCitusVersion(ERROR);
 
+	/*
+	 * Ensure schema exists on each worker node. We can not run this function
+	 * transactionally, since we may create shards over separate sessions and
+	 * shard creation depends on the schema being present and visible from all
+	 * sessions.
+	 */
+	EnsureSchemaExistsOnAllNodes(relationId);
+
 	/*
 	 * Lock target relation with an exclusive lock - there's no way to make
 	 * sense of this table until we've committed, and we don't want multiple

src/test/regress/expected/multi_create_table.out

@@ -873,5 +873,60 @@ SELECT count(*) FROM tt1;
      6
 (1 row)
 
+-- the goal of the following test is to make sure that
+-- both create_reference_table and create_distributed_table
+-- calls creates the schemas without leading to any deadlocks
+-- first create reference table, then hash distributed table
+BEGIN;
+CREATE SCHEMA sc;
+CREATE TABLE sc.ref(a int);
+insert into sc.ref SELECT s FROM generate_series(0, 100) s;
+SELECT create_reference_table('sc.ref');
+NOTICE:  Copying data from local table...
+ create_reference_table 
+------------------------
+ 
+(1 row)
+
+CREATE TABLE sc.hash(a int);
+insert into sc.hash SELECT s FROM generate_series(0, 100) s;
+SELECT create_distributed_table('sc.hash', 'a');
+NOTICE:  Copying data from local table...
+ create_distributed_table 
+--------------------------
+ 
+(1 row)
+
+COMMIT;
+-- first create hash distributed table, then reference table
+BEGIN;
+CREATE SCHEMA sc2;
+CREATE TABLE sc2.hash(a int);
+insert into sc2.hash SELECT s FROM generate_series(0, 100) s;
+SELECT create_distributed_table('sc2.hash', 'a');
+NOTICE:  Copying data from local table...
+ create_distributed_table 
+--------------------------
+ 
+(1 row)
+
+CREATE TABLE sc2.ref(a int);
+insert into sc2.ref SELECT s FROM generate_series(0, 100) s;
+SELECT create_reference_table('sc2.ref');
+NOTICE:  Copying data from local table...
+ create_reference_table 
+------------------------
+ 
+(1 row)
+
+COMMIT;
 DROP TABLE tt1;
 DROP TABLE tt2;
+DROP SCHEMA sc CASCADE;
+NOTICE:  drop cascades to 2 other objects
+DETAIL:  drop cascades to table sc.ref
+drop cascades to table sc.hash
+DROP SCHEMA sc2 CASCADE;
+NOTICE:  drop cascades to 2 other objects
+DETAIL:  drop cascades to table sc2.hash
+drop cascades to table sc2.ref

src/test/regress/sql/multi_create_table.sql

@@ -482,5 +482,41 @@ END;
 
 SELECT count(*) FROM tt1;
 
+
+-- the goal of the following test is to make sure that
+-- both create_reference_table and create_distributed_table
+-- calls creates the schemas without leading to any deadlocks
+
+-- first create reference table, then hash distributed table
+BEGIN;
+
+CREATE SCHEMA sc;
+CREATE TABLE sc.ref(a int);
+insert into sc.ref SELECT s FROM generate_series(0, 100) s;
+SELECT create_reference_table('sc.ref');
+
+CREATE TABLE sc.hash(a int);
+insert into sc.hash SELECT s FROM generate_series(0, 100) s;
+SELECT create_distributed_table('sc.hash', 'a');
+
+COMMIT;
+
+
+-- first create hash distributed table, then reference table
+BEGIN;
+
+CREATE SCHEMA sc2;
+CREATE TABLE sc2.hash(a int);
+insert into sc2.hash SELECT s FROM generate_series(0, 100) s;
+SELECT create_distributed_table('sc2.hash', 'a');
+
+CREATE TABLE sc2.ref(a int);
+insert into sc2.ref SELECT s FROM generate_series(0, 100) s;
+SELECT create_reference_table('sc2.ref');
+
+COMMIT;
+
 DROP TABLE tt1;
 DROP TABLE tt2;
+DROP SCHEMA sc CASCADE;
+DROP SCHEMA sc2 CASCADE;