diff --git a/src/backend/distributed/commands/create_distributed_table.c b/src/backend/distributed/commands/create_distributed_table.c
index 43bd1ebd6..eb51195bf 100644
--- a/src/backend/distributed/commands/create_distributed_table.c
+++ b/src/backend/distributed/commands/create_distributed_table.c
@@ -60,6 +60,7 @@
 #include "distributed/remote_commands.h"
 #include "distributed/shared_library_init.h"
 #include "distributed/worker_protocol.h"
+#include "distributed/worker_shard_visibility.h"
 #include "distributed/worker_transaction.h"
 #include "distributed/version_compat.h"
 #include "executor/executor.h"
@@ -327,6 +328,7 @@ create_reference_table(PG_FUNCTION_ARGS)
  * - we are on the coordinator
  * - the current user is the owner of the table
  * - relation kind is supported
+ * - relation is not a shard
  */
 static void
 EnsureCitusTableCanBeCreated(Oid relationOid)
@@ -343,6 +345,14 @@ EnsureCitusTableCanBeCreated(Oid relationOid)
 	 * will be performed in CreateDistributedTable.
 	 */
 	EnsureRelationKindSupported(relationOid);
+
+	/*
+	 * When coordinator is added to the metadata, or on the workers,
+	 * some of the relations of the coordinator node may/will be shards.
+	 * We disallow creating distributed tables from shard relations, by
+	 * erroring out here.
+	 */
+	ErrorIfRelationIsAKnownShard(relationOid);
 }
 
 
diff --git a/src/test/regress/expected/single_node.out b/src/test/regress/expected/single_node.out
index 0f94e6e7b..3b9f9f2c7 100644
--- a/src/test/regress/expected/single_node.out
+++ b/src/test/regress/expected/single_node.out
@@ -2011,8 +2011,28 @@ RESET citus.enable_manual_changes_to_shards ;
 -- these should work as expected
 TRUNCATE TABLE test_disabling_drop_and_truncate_102040;
 DROP TABLE test_disabling_drop_and_truncate_102040;
-RESET citus.shard_replication_factor;
 DROP TABLE test_disabling_drop_and_truncate;
+-- test creating distributed or reference tables from shards
+CREATE TABLE test_creating_distributed_relation_table_from_shard (a int);
+SELECT create_distributed_table('test_creating_distributed_relation_table_from_shard', 'a');
+NOTICE:  executing the command locally: SELECT worker_apply_shard_ddl_command (102044, 'single_node', 'CREATE TABLE single_node.test_creating_distributed_relation_table_from_shard (a integer) ');SELECT worker_apply_shard_ddl_command (102044, 'single_node', 'ALTER TABLE single_node.test_creating_distributed_relation_table_from_shard OWNER TO postgres')
+NOTICE:  executing the command locally: SELECT worker_apply_shard_ddl_command (102045, 'single_node', 'CREATE TABLE single_node.test_creating_distributed_relation_table_from_shard (a integer) ');SELECT worker_apply_shard_ddl_command (102045, 'single_node', 'ALTER TABLE single_node.test_creating_distributed_relation_table_from_shard OWNER TO postgres')
+NOTICE:  executing the command locally: SELECT worker_apply_shard_ddl_command (102046, 'single_node', 'CREATE TABLE single_node.test_creating_distributed_relation_table_from_shard (a integer) ');SELECT worker_apply_shard_ddl_command (102046, 'single_node', 'ALTER TABLE single_node.test_creating_distributed_relation_table_from_shard OWNER TO postgres')
+NOTICE:  executing the command locally: SELECT worker_apply_shard_ddl_command (102047, 'single_node', 'CREATE TABLE single_node.test_creating_distributed_relation_table_from_shard (a integer) ');SELECT worker_apply_shard_ddl_command (102047, 'single_node', 'ALTER TABLE single_node.test_creating_distributed_relation_table_from_shard OWNER TO postgres')
+ create_distributed_table
+---------------------------------------------------------------------
+
+(1 row)
+
+-- these should error because shards cannot be used to:
+-- create distributed table
+SELECT create_distributed_table('test_creating_distributed_relation_table_from_shard_102044', 'a');
+ERROR:  relation "test_creating_distributed_relation_table_from_shard_102044" is a shard relation
+-- create reference table
+SELECT create_reference_table('test_creating_distributed_relation_table_from_shard_102044');
+ERROR:  relation "test_creating_distributed_relation_table_from_shard_102044" is a shard relation
+RESET citus.shard_replication_factor;
+DROP TABLE test_creating_distributed_relation_table_from_shard;
 -- lets flush the copy often to make sure everyhing is fine
 SET citus.local_copy_flush_threshold TO 1;
 TRUNCATE another_schema_table;
diff --git a/src/test/regress/expected/single_node_truncate.out b/src/test/regress/expected/single_node_truncate.out
index bf1c99d69..7d94dc744 100644
--- a/src/test/regress/expected/single_node_truncate.out
+++ b/src/test/regress/expected/single_node_truncate.out
@@ -31,9 +31,9 @@ SELECT * FROM table_sizes;
         name        | has_data
 ---------------------------------------------------------------------
  citus_local        | f
- citus_local_102045 | t
+ citus_local_102049 | t
  ref                | t
- ref_102044         | t
+ ref_102048         | t
 (4 rows)
 
 -- verify that this UDF is noop on Citus local tables
@@ -47,9 +47,9 @@ SELECT * FROM table_sizes;
         name        | has_data
 ---------------------------------------------------------------------
  citus_local        | f
- citus_local_102045 | t
+ citus_local_102049 | t
  ref                | t
- ref_102044         | t
+ ref_102048         | t
 (4 rows)
 
 -- test that we allow cascading truncates to citus local tables
@@ -65,9 +65,9 @@ SELECT * FROM table_sizes;
         name        | has_data
 ---------------------------------------------------------------------
  citus_local        | f
- citus_local_102045 | t
+ citus_local_102049 | t
  ref                | f
- ref_102044         | t
+ ref_102048         | t
 (4 rows)
 
 ROLLBACK;
@@ -98,14 +98,14 @@ SELECT * FROM table_sizes;
         name        | has_data
 ---------------------------------------------------------------------
  citus_local        | f
- citus_local_102045 | t
+ citus_local_102049 | t
  dist               | f
- dist_102047        | t
- dist_102048        | t
- dist_102049        | t
- dist_102050        | t
+ dist_102051        | t
+ dist_102052        | t
+ dist_102053        | t
+ dist_102054        | t
  ref                | f
- ref_102044         | t
+ ref_102048         | t
 (9 rows)
 
 ROLLBACK;
@@ -121,14 +121,14 @@ SELECT * FROM table_sizes;
         name        | has_data
 ---------------------------------------------------------------------
  citus_local        | f
- citus_local_102045 | t
+ citus_local_102049 | t
  dist               | f
- dist_102047        | t
- dist_102048        | t
- dist_102049        | t
- dist_102050        | t
+ dist_102051        | t
+ dist_102052        | t
+ dist_102053        | t
+ dist_102054        | t
  ref                | t
- ref_102044         | t
+ ref_102048         | t
 (9 rows)
 
 ROLLBACK;
diff --git a/src/test/regress/sql/single_node.sql b/src/test/regress/sql/single_node.sql
index 13659f2eb..3ca456108 100644
--- a/src/test/regress/sql/single_node.sql
+++ b/src/test/regress/sql/single_node.sql
@@ -1017,9 +1017,22 @@ RESET citus.enable_manual_changes_to_shards ;
 TRUNCATE TABLE test_disabling_drop_and_truncate_102040;
 DROP TABLE test_disabling_drop_and_truncate_102040;
 
-RESET citus.shard_replication_factor;
 DROP TABLE test_disabling_drop_and_truncate;
 
+-- test creating distributed or reference tables from shards
+CREATE TABLE test_creating_distributed_relation_table_from_shard (a int);
+SELECT create_distributed_table('test_creating_distributed_relation_table_from_shard', 'a');
+
+-- these should error because shards cannot be used to:
+-- create distributed table
+SELECT create_distributed_table('test_creating_distributed_relation_table_from_shard_102044', 'a');
+
+-- create reference table
+SELECT create_reference_table('test_creating_distributed_relation_table_from_shard_102044');
+
+RESET citus.shard_replication_factor;
+DROP TABLE test_creating_distributed_relation_table_from_shard;
+
 -- lets flush the copy often to make sure everyhing is fine
 SET citus.local_copy_flush_threshold TO 1;
 TRUNCATE another_schema_table;