Merge pull request #5836 from citusdata/fix/cannot-distribute-tmp-schemas

src/backend/distributed/commands/collation.c

@@ -19,6 +19,7 @@
 #include "distributed/deparser.h"
 #include "distributed/listutils.h"
 #include "distributed/metadata_utility.h"
+#include "distributed/metadata/dependency.h"
 #include "distributed/metadata/distobject.h"
 #include "distributed/metadata_sync.h"
 #include "distributed/multi_executor.h"
@@ -588,6 +589,14 @@ PostprocessDefineCollationStmt(Node *node, const char *queryString)
 	ObjectAddress collationAddress =
 		DefineCollationStmtObjectAddress(node, false);
 
+	DeferredErrorMessage *errMsg = DeferErrorIfHasUnsupportedDependency(
+		&collationAddress);
+	if (errMsg != NULL)
+	{
+		RaiseDeferredError(errMsg, WARNING);
+		return NIL;
+	}
+
 	EnsureDependenciesExistOnAllNodes(&collationAddress);
 
 	/* to prevent recursion with mx we disable ddl propagation */

src/backend/distributed/commands/dependencies.c

@@ -158,6 +158,8 @@ EnsureDependenciesCanBeDistributed(const ObjectAddress *objectAddress)
 
 	if (depError != NULL)
 	{
+		/* override error detail as it is not applicable here*/
+		depError->detail = NULL;
 		RaiseDeferredError(depError, ERROR);
 	}
 }

src/backend/distributed/commands/text_search.c

@@ -34,6 +34,7 @@
 #include "distributed/commands/utility_hook.h"
 #include "distributed/deparser.h"
 #include "distributed/listutils.h"
+#include "distributed/metadata/dependency.h"
 #include "distributed/metadata/distobject.h"
 #include "distributed/metadata_sync.h"
 #include "distributed/multi_executor.h"
@@ -91,6 +92,14 @@ PostprocessCreateTextSearchConfigurationStmt(Node *node, const char *queryString
 	EnsureSequentialMode(OBJECT_TSCONFIGURATION);
 
 	ObjectAddress address = GetObjectAddressFromParseTree((Node *) stmt, false);
+
+	DeferredErrorMessage *errMsg = DeferErrorIfHasUnsupportedDependency(&address);
+	if (errMsg != NULL)
+	{
+		RaiseDeferredError(errMsg, WARNING);
+		return NIL;
+	}
+
 	EnsureDependenciesExistOnAllNodes(&address);
 
 	/*
@@ -132,6 +141,14 @@ PostprocessCreateTextSearchDictionaryStmt(Node *node, const char *queryString)
 	EnsureSequentialMode(OBJECT_TSDICTIONARY);
 
 	ObjectAddress address = GetObjectAddressFromParseTree((Node *) stmt, false);
+
+	DeferredErrorMessage *errMsg = DeferErrorIfHasUnsupportedDependency(&address);
+	if (errMsg != NULL)
+	{
+		RaiseDeferredError(errMsg, WARNING);
+		return NIL;
+	}
+
 	EnsureDependenciesExistOnAllNodes(&address);
 
 	QualifyTreeNode(node);

src/backend/distributed/commands/type.c

@@ -260,22 +260,7 @@ PreprocessCreateEnumStmt(Node *node, const char *queryString,
 	/* enforce fully qualified typeName for correct deparsing and lookup */
 	QualifyTreeNode(node);
 
-	/* reconstruct creation statement in a portable fashion */
+	return NIL;
-	const char *createEnumStmtSql = DeparseCreateEnumStmt(node);
-	createEnumStmtSql = WrapCreateOrReplace(createEnumStmtSql);
-
-	/*
-	 * when we allow propagation within a transaction block we should make sure to only
-	 * allow this in sequential mode
-	 */
-	EnsureSequentialMode(OBJECT_TYPE);
-
-	/* to prevent recursion with mx we disable ddl propagation */
-	List *commands = list_make3(DISABLE_DDL_PROPAGATION,
-								(void *) createEnumStmtSql,
-								ENABLE_DDL_PROPAGATION);
-
-	return NodeDDLTaskList(NON_COORDINATOR_NODES, commands);
 }
 
 
@@ -297,9 +282,32 @@ PostprocessCreateEnumStmt(Node *node, const char *queryString)
 
 	/* lookup type address of just created type */
 	ObjectAddress typeAddress = GetObjectAddressFromParseTree(node, false);
+
+	DeferredErrorMessage *errMsg = DeferErrorIfHasUnsupportedDependency(&typeAddress);
+	if (errMsg != NULL)
+	{
+		RaiseDeferredError(errMsg, WARNING);
+		return NIL;
+	}
+
+	/*
+	 * when we allow propagation within a transaction block we should make sure to only
+	 * allow this in sequential mode
+	 */
+	EnsureSequentialMode(OBJECT_TYPE);
+
 	EnsureDependenciesExistOnAllNodes(&typeAddress);
 
-	return NIL;
+	/* reconstruct creation statement in a portable fashion */
+	const char *createEnumStmtSql = DeparseCreateEnumStmt(node);
+	createEnumStmtSql = WrapCreateOrReplace(createEnumStmtSql);
+
+	/* to prevent recursion with mx we disable ddl propagation */
+	List *commands = list_make3(DISABLE_DDL_PROPAGATION,
+								(void *) createEnumStmtSql,
+								ENABLE_DDL_PROPAGATION);
+
+	return NodeDDLTaskList(NON_COORDINATOR_NODES, commands);
 }
 
 

src/backend/distributed/metadata/dependency.c

@@ -600,7 +600,7 @@ SupportedDependencyByCitus(const ObjectAddress *address)
 		{
 			case OCLASS_SCHEMA:
 			{
-				return true;
+				return !isTempNamespace(address->objectId);
 			}
 
 			default:
@@ -631,11 +631,15 @@ SupportedDependencyByCitus(const ObjectAddress *address)
 		}
 
 		case OCLASS_COLLATION:
-		case OCLASS_SCHEMA:
 		{
 			return true;
 		}
 
+		case OCLASS_SCHEMA:
+		{
+			return !isTempNamespace(address->objectId);
+		}
+
 		case OCLASS_PROC:
 		{
 			return true;
@@ -776,15 +780,16 @@ DeferErrorIfHasUnsupportedDependency(const ObjectAddress *objectAddress)
 	#endif
 
 	/*
-	 * If the given object is a procedure or type, we want to create it locally,
+	 * We expect callers to interpret the error returned from this function
-	 * so provide that information in the error detail.
+	 * as a warning if the object itself is just being created. In that case,
+	 * we expect them to report below error detail as well to indicate that
+	 * object itself will not be propagated but will still be created locally.
+	 *
+	 * Otherwise, callers are expected to throw the error returned from this
+	 * function as a hard one by ignoring the detail part.
 	 */
-	if (getObjectClass(objectAddress) == OCLASS_PROC ||
+	appendStringInfo(detailInfo, "\"%s\" will be created only locally",
-		getObjectClass(objectAddress) == OCLASS_TYPE)
+					 objectDescription);
-	{
-		appendStringInfo(detailInfo, "\"%s\" will be created only locally",
-						 objectDescription);
-	}
 
 	if (SupportedDependencyByCitus(undistributableDependency))
 	{
@@ -800,9 +805,7 @@ DeferErrorIfHasUnsupportedDependency(const ObjectAddress *objectAddress)
 						 objectDescription);
 
 		return DeferredError(ERRCODE_FEATURE_NOT_SUPPORTED,
-							 errorInfo->data,
+							 errorInfo->data, detailInfo->data, hintInfo->data);
-							 strlen(detailInfo->data) == 0 ? NULL : detailInfo->data,
-							 hintInfo->data);
 	}
 
 	appendStringInfo(errorInfo, "\"%s\" has dependency on unsupported "
@@ -810,9 +813,7 @@ DeferErrorIfHasUnsupportedDependency(const ObjectAddress *objectAddress)
 					 dependencyDescription);
 
 	return DeferredError(ERRCODE_FEATURE_NOT_SUPPORTED,
-						 errorInfo->data,
+						 errorInfo->data, detailInfo->data, NULL);
-						 strlen(detailInfo->data) == 0 ? NULL : detailInfo->data,
-						 NULL);
 }
 
 

src/test/regress/bin/normalize.sed

@@ -263,3 +263,23 @@ s/issuing SELECT pg_cancel_backend\([0-9]+::integer\)/issuing SELECT pg_cancel_b
 
 # node id in run_command_on_all_nodes warning
 s/Error on node with node id [0-9]+/Error on node with node id xxxxx/g
+
+# Temp schema names in error messages regarding dependencies that we cannot distribute
+#
+# 1) Schema of the depending object in the error message:
+#
+# e.g.:
+#   WARNING:  "function pg_temp_3.f(bigint)" has dependency on unsupported object "<foo>"
+# will be replaced with
+#   WARNING:  "function pg_temp_xxx.f(bigint)" has dependency on unsupported object "<foo>"
+s/^(WARNING|ERROR)(:  "[a-z\ ]+ )pg_temp_[0-9]+(\..*" has dependency on unsupported object ".*")$/\1\2pg_temp_xxx\3/g
+
+# 2) Schema of the depending object in the error detail:
+s/^(DETAIL:  "[a-z\ ]+ )pg_temp_[0-9]+(\..*" will be created only locally)$/\1pg_temp_xxx\2/g
+
+# 3) Schema that the object depends in the error message:
+# e.g.:
+#   WARNING:  "function func(bigint)" has dependency on unsupported object "schema pg_temp_3"
+# will be replaced with
+#   WARNING:  "function func(bigint)" has dependency on unsupported object "schema pg_temp_xxx"
+s/^(WARNING|ERROR)(:  "[a-z\ ]+ .*" has dependency on unsupported object) "schema pg_temp_[0-9]+"$/\1\2 "schema pg_temp_xxx"/g

src/test/regress/expected/distributed_collations.out

@@ -179,3 +179,8 @@ HINT:  Connect to the coordinator and run it again.
 SET citus.enable_ddl_propagation TO off;
 DROP SCHEMA collation_creation_on_worker;
 SET citus.enable_ddl_propagation TO on;
+\c - - - :master_port
+-- will skip trying to propagate the collation due to temp schema
+CREATE COLLATION pg_temp.temp_collation (provider = icu, locale = 'de-u-co-phonebk');
+WARNING:  "collation pg_temp_xxx.temp_collation" has dependency on unsupported object "schema pg_temp_xxx"
+DETAIL:  "collation pg_temp_xxx.temp_collation" will be created only locally

src/test/regress/expected/distributed_functions.out

@@ -80,6 +80,33 @@ CREATE FUNCTION add_polygons(polygon, polygon) RETURNS int
     LANGUAGE SQL
     IMMUTABLE
     RETURNS NULL ON NULL INPUT;
+CREATE FUNCTION agg_dummy_func(state int, item int)
+RETURNS int IMMUTABLE LANGUAGE plpgsql AS $$
+begin
+    return state + item;
+end;
+$$;
+SET client_min_messages TO WARNING;
+-- will skip trying to propagate the aggregate due to temp schema
+CREATE AGGREGATE pg_temp.dummy_agg(int) (
+    sfunc = agg_dummy_func,
+    stype = int,
+    sspace = 8,
+    finalfunc = agg_dummy_func,
+    finalfunc_extra,
+    initcond = '5',
+    msfunc = agg_dummy_func,
+    mstype = int,
+    msspace = 12,
+    minvfunc = agg_dummy_func,
+    mfinalfunc = agg_dummy_func,
+    mfinalfunc_extra,
+    minitcond = '1',
+    sortop = ">"
+);
+WARNING:  "function pg_temp_xxx.dummy_agg(integer)" has dependency on unsupported object "schema pg_temp_xxx"
+DETAIL:  "function pg_temp_xxx.dummy_agg(integer)" will be created only locally
+RESET client_min_messages;
 -- Test some combination of functions without ddl propagation
 -- This will prevent the workers from having those types created. They are
 -- created just-in-time on function distribution

src/test/regress/expected/distributed_types.out

@@ -597,6 +597,13 @@ DETAIL:  "type default_test_row" will be created only locally
 CREATE TABLE table_text_local_def(id int, col_1 default_test_row);
 SELECT create_distributed_table('table_text_local_def','id');
 ERROR:  "table table_text_local_def" has dependency on unsupported object "type text_local_def"
+-- will skip trying to propagate the type/enum due to temp schema
+CREATE TYPE pg_temp.temp_type AS (int_field int);
+WARNING:  "type temp_type" has dependency on unsupported object "schema pg_temp_xxx"
+DETAIL:  "type temp_type" will be created only locally
+CREATE TYPE pg_temp.temp_enum AS ENUM ('one', 'two', 'three');
+WARNING:  "type temp_enum" has dependency on unsupported object "schema pg_temp_xxx"
+DETAIL:  "type temp_enum" will be created only locally
 -- clear objects
 SET client_min_messages TO error; -- suppress cascading objects dropping
 DROP SCHEMA type_tests CASCADE;

src/test/regress/expected/function_propagation.out

@@ -1279,6 +1279,17 @@ ALTER TABLE table_1_for_circ_dep_3 ADD COLUMN col_2 table_2_for_circ_dep_3;
 SELECT create_distributed_table('table_1_for_circ_dep_3','id');
 ERROR:  Citus can not handle circular dependencies between distributed objects
 DETAIL:  "table table_1_for_circ_dep_3" circularly depends itself, resolve circular dependency first
+-- will skip trying to propagate the function due to temp schema
+CREATE FUNCTION pg_temp.temp_func(group_size BIGINT) RETURNS SETOF integer[]
+AS $$
+    SELECT array_agg(s) OVER w
+      FROM generate_series(1,5) s
+    WINDOW w AS (ORDER BY s ROWS BETWEEN CURRENT ROW AND GROUP_SIZE FOLLOWING)
+$$ LANGUAGE SQL STABLE;
+WARNING:  "function pg_temp_xxx.temp_func(bigint)" has dependency on unsupported object "schema pg_temp_xxx"
+DETAIL:  "function pg_temp_xxx.temp_func(bigint)" will be created only locally
+SELECT create_distributed_function('pg_temp.temp_func(BIGINT)');
+ERROR:  "function pg_temp_xxx.temp_func(bigint)" has dependency on unsupported object "schema pg_temp_xxx"
 RESET search_path;
 SET client_min_messages TO WARNING;
 DROP SCHEMA function_propagation_schema CASCADE;

src/test/regress/expected/multi_table_ddl.out

@@ -163,6 +163,13 @@ ALTER TABLE test_table ADD COLUMN id3 bigserial;
 ERROR:  cannot execute ADD COLUMN commands involving serial pseudotypes when metadata is synchronized to workers
 ALTER TABLE test_table ADD COLUMN id4 bigserial CHECK (id4 > 0);
 ERROR:  cannot execute ADD COLUMN commands involving serial pseudotypes when metadata is synchronized to workers
+CREATE SEQUENCE pg_temp.temp_sequence;
+CREATE TABLE table_with_temp_sequence (
+  dist_key int,
+  seq_col bigint default nextval('pg_temp.temp_sequence')
+);
+SELECT create_distributed_table('table_with_temp_sequence', 'dist_key');
+ERROR:  "table table_with_temp_sequence" has dependency on unsupported object "schema pg_temp_xxx"
 DROP TABLE test_table CASCADE;
 DROP SEQUENCE test_sequence_0;
 DROP SEQUENCE test_sequence_1;

src/test/regress/expected/propagate_extension_commands.out

@@ -628,5 +628,10 @@ objid = (SELECT oid FROM pg_proc WHERE prosrc = 'cube_a_f8_f8');
 (1 row)
 
 ROLLBACK;
+-- Postgres already doesn't allow creating extensions in temp schema but
+-- let's have a test for that to track any furher changes in postgres.
+DROP EXTENSION isn CASCADE;
+CREATE EXTENSION isn WITH SCHEMA pg_temp;
+ERROR:  schema "pg_temp" does not exist
 -- drop the schema and all the objects
 DROP SCHEMA "extension'test" CASCADE;

src/test/regress/expected/propagate_extension_commands_1.out

@@ -634,5 +634,10 @@ objid = (SELECT oid FROM pg_proc WHERE prosrc = 'cube_a_f8_f8');
 (1 row)
 
 ROLLBACK;
+-- Postgres already doesn't allow creating extensions in temp schema but
+-- let's have a test for that to track any furher changes in postgres.
+DROP EXTENSION isn CASCADE;
+CREATE EXTENSION isn WITH SCHEMA pg_temp;
+ERROR:  schema "pg_temp" does not exist
 -- drop the schema and all the objects
 DROP SCHEMA "extension'test" CASCADE;

src/test/regress/expected/propagate_statistics.out

@@ -15,6 +15,8 @@ SELECT create_distributed_table('test_stats', 'a');
 
 (1 row)
 
+CREATE STATISTICS pg_temp.s1 (dependencies) ON a, b FROM test_stats;
+ERROR:  "statistics object s1" has dependency on unsupported object "schema pg_temp_xxx"
 CREATE STATISTICS s1 (dependencies) ON a, b FROM test_stats;
 -- test for distributing an already existing statistics
 CREATE TABLE "test'stats2" (

src/test/regress/expected/text_search.out

@@ -696,6 +696,18 @@ $$);
  t
 (1 row)
 
+-- will skip trying to propagate the text search configuration due to temp schema
+CREATE TEXT SEARCH CONFIGURATION pg_temp.temp_text_search_config ( parser = default );
+WARNING:  "text search configuration pg_temp_xxx.temp_text_search_config" has dependency on unsupported object "schema pg_temp_xxx"
+DETAIL:  "text search configuration pg_temp_xxx.temp_text_search_config" will be created only locally
+-- will skip trying to propagate the text search dictionary due to temp schema
+CREATE TEXT SEARCH DICTIONARY pg_temp.temp_text_search_dict (
+    template = snowball,
+    language = english,
+    stopwords = english
+);
+WARNING:  "text search dictionary pg_temp_xxx.temp_text_search_dict" has dependency on unsupported object "schema pg_temp_xxx"
+DETAIL:  "text search dictionary pg_temp_xxx.temp_text_search_dict" will be created only locally
 SET client_min_messages TO 'warning';
 DROP SCHEMA text_search, text_search2, "Text Search Requiring Quote's" CASCADE;
 DROP ROLE text_search_owner;

src/test/regress/sql/distributed_collations.sql

@@ -109,3 +109,8 @@ CREATE COLLATION collation_creation_on_worker.another_german_phonebook (provider
 SET citus.enable_ddl_propagation TO off;
 DROP SCHEMA collation_creation_on_worker;
 SET citus.enable_ddl_propagation TO on;
+
+\c - - - :master_port
+
+-- will skip trying to propagate the collation due to temp schema
+CREATE COLLATION pg_temp.temp_collation (provider = icu, locale = 'de-u-co-phonebk');

src/test/regress/sql/distributed_functions.sql

@@ -61,6 +61,33 @@ CREATE FUNCTION add_polygons(polygon, polygon) RETURNS int
     IMMUTABLE
     RETURNS NULL ON NULL INPUT;
 
+CREATE FUNCTION agg_dummy_func(state int, item int)
+RETURNS int IMMUTABLE LANGUAGE plpgsql AS $$
+begin
+    return state + item;
+end;
+$$;
+
+SET client_min_messages TO WARNING;
+-- will skip trying to propagate the aggregate due to temp schema
+CREATE AGGREGATE pg_temp.dummy_agg(int) (
+    sfunc = agg_dummy_func,
+    stype = int,
+    sspace = 8,
+    finalfunc = agg_dummy_func,
+    finalfunc_extra,
+    initcond = '5',
+    msfunc = agg_dummy_func,
+    mstype = int,
+    msspace = 12,
+    minvfunc = agg_dummy_func,
+    mfinalfunc = agg_dummy_func,
+    mfinalfunc_extra,
+    minitcond = '1',
+    sortop = ">"
+);
+RESET client_min_messages;
+
 -- Test some combination of functions without ddl propagation
 -- This will prevent the workers from having those types created. They are
 -- created just-in-time on function distribution

src/test/regress/sql/distributed_types.sql

@@ -359,6 +359,10 @@ CREATE TYPE default_test_row AS (f1 text_local_def, f2 int4);
 CREATE TABLE table_text_local_def(id int, col_1 default_test_row);
 SELECT create_distributed_table('table_text_local_def','id');
 
+-- will skip trying to propagate the type/enum due to temp schema
+CREATE TYPE pg_temp.temp_type AS (int_field int);
+CREATE TYPE pg_temp.temp_enum AS ENUM ('one', 'two', 'three');
+
 -- clear objects
 SET client_min_messages TO error; -- suppress cascading objects dropping
 DROP SCHEMA type_tests CASCADE;

src/test/regress/sql/function_propagation.sql

@@ -832,6 +832,16 @@ ALTER TABLE table_1_for_circ_dep_3 ADD COLUMN col_2 table_2_for_circ_dep_3;
 -- It should error out due to circular dependency
 SELECT create_distributed_table('table_1_for_circ_dep_3','id');
 
+-- will skip trying to propagate the function due to temp schema
+CREATE FUNCTION pg_temp.temp_func(group_size BIGINT) RETURNS SETOF integer[]
+AS $$
+    SELECT array_agg(s) OVER w
+      FROM generate_series(1,5) s
+    WINDOW w AS (ORDER BY s ROWS BETWEEN CURRENT ROW AND GROUP_SIZE FOLLOWING)
+$$ LANGUAGE SQL STABLE;
+
+SELECT create_distributed_function('pg_temp.temp_func(BIGINT)');
+
 RESET search_path;
 SET client_min_messages TO WARNING;
 DROP SCHEMA function_propagation_schema CASCADE;

src/test/regress/sql/multi_table_ddl.sql

@@ -122,6 +122,13 @@ ALTER TABLE test_table ALTER COLUMN id3 SET DEFAULT nextval('test_sequence_1'),
 ALTER TABLE test_table ADD COLUMN id3 bigserial;
 ALTER TABLE test_table ADD COLUMN id4 bigserial CHECK (id4 > 0);
 
+CREATE SEQUENCE pg_temp.temp_sequence;
+CREATE TABLE table_with_temp_sequence (
+  dist_key int,
+  seq_col bigint default nextval('pg_temp.temp_sequence')
+);
+SELECT create_distributed_table('table_with_temp_sequence', 'dist_key');
+
 DROP TABLE test_table CASCADE;
 DROP SEQUENCE test_sequence_0;
 DROP SEQUENCE test_sequence_1;

src/test/regress/sql/propagate_extension_commands.sql

@@ -371,5 +371,10 @@ SELECT distribution_argument_index FROM pg_catalog.pg_dist_object WHERE classid
 objid = (SELECT oid FROM pg_proc WHERE prosrc = 'cube_a_f8_f8');
 ROLLBACK;
 
+-- Postgres already doesn't allow creating extensions in temp schema but
+-- let's have a test for that to track any furher changes in postgres.
+DROP EXTENSION isn CASCADE;
+CREATE EXTENSION isn WITH SCHEMA pg_temp;
+
 -- drop the schema and all the objects
 DROP SCHEMA "extension'test" CASCADE;

src/test/regress/sql/propagate_statistics.sql

@@ -14,6 +14,8 @@ CREATE TABLE test_stats (
 
 SELECT create_distributed_table('test_stats', 'a');
 
+CREATE STATISTICS pg_temp.s1 (dependencies) ON a, b FROM test_stats;
+
 CREATE STATISTICS s1 (dependencies) ON a, b FROM test_stats;
 
 -- test for distributing an already existing statistics

src/test/regress/sql/text_search.sql

@@ -402,6 +402,16 @@ SELECT COUNT(DISTINCT result)=1 FROM run_command_on_all_nodes($$
     WHERE dictname = 'snowball_dict';
 $$);
 
+-- will skip trying to propagate the text search configuration due to temp schema
+CREATE TEXT SEARCH CONFIGURATION pg_temp.temp_text_search_config ( parser = default );
+
+-- will skip trying to propagate the text search dictionary due to temp schema
+CREATE TEXT SEARCH DICTIONARY pg_temp.temp_text_search_dict (
+    template = snowball,
+    language = english,
+    stopwords = english
+);
+
 SET client_min_messages TO 'warning';
 DROP SCHEMA text_search, text_search2, "Text Search Requiring Quote's" CASCADE;
 DROP ROLE text_search_owner;