Prevent Citus table functions from being called on shards (Fix #5610) (#5694)

DESCRIPTION: Prevent Citus table functions from being called on shards

The operations that guard against using shards are:
* Create Local Table
* Create distributed table (which affects reference table creation as well).

* I used a `ErrorIfRaltionIsKnownShard` instead of `ErrorIfIllegallyChangingKnownShard`.
`ErrorIfIllegallyChangingKnownShard` allows the operation if `citus.enable_manual_changes_to_shards`,
but I am not sure if it ever makes sense to create a distributed, reference, or citus local table out of a shard.

I tried to go over the code to identify other UDF-s where shards could be illegaly changed, but I could not find any other.
My knowledge of the codebase is not solid enough for me to say for sure.

Fixes #5610

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);
 }
 
 

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;

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_102051        | t
- dist_102048        | t
+ dist_102052        | t
- dist_102049        | t
+ dist_102053        | t
- dist_102050        | 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_102051        | t
- dist_102048        | t
+ dist_102052        | t
- dist_102049        | t
+ dist_102053        | t
- dist_102050        | t
+ dist_102054        | t
  ref                | t
- ref_102044         | t
+ ref_102048         | t
 (9 rows)
 
 ROLLBACK;

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;