Merge pull request #5893 from citusdata/velioglu/fix_function_in_tx

Create function in transaction according to create object propagation guc

src/backend/distributed/commands/function.c

@@ -1263,12 +1263,7 @@ ShouldPropagateCreateFunction(CreateFunctionStmt *stmt)
 		return false;
 	}
 
-	/*
+	if (!ShouldPropagateCreateInCoordinatedTransction())
-	 * If the create command is a part of a multi-statement transaction that is not in
-	 * sequential mode, don't propagate.
-	 */
-	if (IsMultiStatementTransaction() &&
-		MultiShardConnectionType != SEQUENTIAL_CONNECTION)
 	{
 		return false;
 	}

src/test/regress/expected/citus_local_table_triggers.out

@@ -33,18 +33,18 @@ NOTICE:  executing the command locally: SELECT worker_apply_inter_shard_ddl_comm
 ---------------------------------------------------------------------
 BEGIN;
     CREATE TABLE distributed_table(value int);
-    SELECT create_distributed_table('distributed_table', 'value');
- create_distributed_table
----------------------------------------------------------------------
-
-(1 row)
-
     CREATE FUNCTION insert_42() RETURNS trigger AS $insert_42$
     BEGIN
         INSERT INTO distributed_table VALUES (42);
         RETURN NEW;
     END;
     $insert_42$ LANGUAGE plpgsql;
+    SELECT create_distributed_table('distributed_table', 'value');
+ create_distributed_table
+---------------------------------------------------------------------
+
+(1 row)
+
     CREATE TRIGGER insert_42_trigger
     AFTER DELETE ON citus_local_table
     FOR EACH ROW EXECUTE FUNCTION insert_42();

src/test/regress/expected/function_propagation.out

@@ -321,6 +321,7 @@ $$;
 -- Show that functions are propagated (or not) as a dependency
 -- Function as a default column
 BEGIN;
+    SET LOCAL citus.create_object_propagation TO deferred;
     CREATE OR REPLACE FUNCTION func_in_transaction_def()
     RETURNS int
     LANGUAGE plpgsql AS
@@ -329,7 +330,6 @@ BEGIN;
         return 1;
     END;
     $$;
-    -- Function shouldn't be propagated within transaction
     SELECT pg_identify_object_as_address(classid, objid, objsubid) from pg_catalog.pg_dist_object where objid = 'function_propagation_schema.func_in_transaction_def'::regproc::oid;
  pg_identify_object_as_address
 ---------------------------------------------------------------------
@@ -360,6 +360,7 @@ SELECT * FROM run_command_on_workers($$SELECT pg_identify_object_as_address(clas
 
 -- Multiple functions as a default column
 BEGIN;
+    SET LOCAL citus.create_object_propagation TO deferred;
     CREATE OR REPLACE FUNCTION func_in_transaction_1()
     RETURNS int
     LANGUAGE plpgsql AS
@@ -376,7 +377,6 @@ BEGIN;
         return 1;
     END;
     $$;
-    -- Functions shouldn't be propagated within transaction
     SELECT pg_identify_object_as_address(classid, objid, objsubid) from pg_catalog.pg_dist_object where objid = 'function_propagation_schema.func_in_transaction_1'::regproc::oid;
  pg_identify_object_as_address
 ---------------------------------------------------------------------
@@ -434,6 +434,9 @@ BEGIN;
         return 1;
     END;
     $$;
+WARNING:  "function func_in_transaction_3(non_dist_table)" has dependency to "table non_dist_table" that is not in Citus' metadata
+DETAIL:  "function func_in_transaction_3(non_dist_table)" will be created only locally
+HINT:  Distribute "table non_dist_table" first to distribute "function func_in_transaction_3(non_dist_table)"
     CREATE TABLE table_to_prop_func_3(id int, col_1 int default func_in_transaction_3(NULL::non_dist_table));
     -- It should error out as there is a non-distributed table dependency
     SELECT create_distributed_table('table_to_prop_func_3','id');
@@ -443,12 +446,7 @@ COMMIT;
 -- Adding a column with default value should propagate the function
 BEGIN;
     CREATE TABLE table_to_prop_func_4(id int);
-    SELECT create_distributed_table('table_to_prop_func_4', 'id');
+    SET LOCAL citus.create_object_propagation TO deferred;
- create_distributed_table
----------------------------------------------------------------------
-
-(1 row)
-
     CREATE OR REPLACE FUNCTION func_in_transaction_4()
     RETURNS int
     LANGUAGE plpgsql AS
@@ -457,7 +455,12 @@ BEGIN;
         return 1;
     END;
     $$;
-    -- Function shouldn't be propagated within transaction
+    SELECT create_distributed_table('table_to_prop_func_4', 'id');
+ create_distributed_table
+---------------------------------------------------------------------
+
+(1 row)
+
     SELECT pg_identify_object_as_address(classid, objid, objsubid) from pg_catalog.pg_dist_object where objid = 'function_propagation_schema.func_in_transaction_4'::regproc::oid;
  pg_identify_object_as_address
 ---------------------------------------------------------------------
@@ -483,6 +486,7 @@ SELECT * FROM run_command_on_workers($$SELECT pg_identify_object_as_address(clas
 -- Adding a column with default function depending on non-distributable table should fail
 BEGIN;
     CREATE TABLE non_dist_table_for_function(id int);
+    SET LOCAL citus.create_object_propagation TO deferred;
     CREATE OR REPLACE FUNCTION non_dist_func(col_1 non_dist_table_for_function)
     RETURNS int
     LANGUAGE plpgsql AS
@@ -504,6 +508,7 @@ HINT:  Distribute "table non_dist_table_for_function" first to distribute "table
 ROLLBACK;
 -- Adding multiple columns with default values should propagate the function
 BEGIN;
+    SET LOCAL citus.create_object_propagation TO deferred;
     CREATE OR REPLACE FUNCTION func_in_transaction_5()
     RETURNS int
     LANGUAGE plpgsql AS
@@ -520,7 +525,6 @@ BEGIN;
         return 1;
     END;
     $$;
-    -- Functions shouldn't be propagated within transaction
     SELECT pg_identify_object_as_address(classid, objid, objsubid) from pg_catalog.pg_dist_object where objid = 'function_propagation_schema.func_in_transaction_5'::regproc::oid;
  pg_identify_object_as_address
 ---------------------------------------------------------------------
@@ -569,6 +573,7 @@ SELECT * FROM run_command_on_workers($$SELECT pg_identify_object_as_address(clas
 
 -- Adding a constraint with function check should propagate the function
 BEGIN;
+    SET LOCAL citus.create_object_propagation TO deferred;
     CREATE OR REPLACE FUNCTION func_in_transaction_7(param_1 int)
     RETURNS boolean
     LANGUAGE plpgsql AS
@@ -577,7 +582,6 @@ BEGIN;
         return param_1 > 5;
     END;
     $$;
-    -- Functions shouldn't be propagated within transaction
     SELECT pg_identify_object_as_address(classid, objid, objsubid) from pg_catalog.pg_dist_object where objid = 'function_propagation_schema.func_in_transaction_7'::regproc::oid;
  pg_identify_object_as_address
 ---------------------------------------------------------------------
@@ -608,6 +612,7 @@ SELECT * FROM run_command_on_workers($$SELECT pg_identify_object_as_address(clas
 
 -- Adding a constraint with multiple functions check should propagate the function
 BEGIN;
+    SET LOCAL citus.create_object_propagation TO deferred;
     CREATE OR REPLACE FUNCTION func_in_transaction_8(param_1 int)
     RETURNS boolean
     LANGUAGE plpgsql AS
@@ -624,7 +629,6 @@ BEGIN;
         return param_1 > 5;
     END;
     $$;
-    -- Functions shouldn't be propagated within transaction
     SELECT pg_identify_object_as_address(classid, objid, objsubid) from pg_catalog.pg_dist_object where objid = 'function_propagation_schema.func_in_transaction_8'::regproc::oid;
  pg_identify_object_as_address
 ---------------------------------------------------------------------
@@ -680,6 +684,7 @@ BEGIN;
 
 (1 row)
 
+    SET LOCAL citus.create_object_propagation TO deferred;
     CREATE OR REPLACE FUNCTION func_in_transaction_10(param_1 int)
     RETURNS boolean
     LANGUAGE plpgsql AS
@@ -688,7 +693,6 @@ BEGIN;
         return param_1 > 5;
     END;
     $$;
-    -- Functions shouldn't be propagated within transaction
     SELECT pg_identify_object_as_address(classid, objid, objsubid) from pg_catalog.pg_dist_object where objid = 'function_propagation_schema.func_in_transaction_10'::regproc::oid;
  pg_identify_object_as_address
 ---------------------------------------------------------------------
@@ -722,6 +726,9 @@ BEGIN;
         return param_1 > 5;
     END;
     $$;
+WARNING:  "function func_in_transaction_11(integer,local_table_for_const)" has dependency to "table local_table_for_const" that is not in Citus' metadata
+DETAIL:  "function func_in_transaction_11(integer,local_table_for_const)" will be created only locally
+HINT:  Distribute "table local_table_for_const" first to distribute "function func_in_transaction_11(integer,local_table_for_const)"
     CREATE TABLE table_to_prop_func_9(id int, col_1 int check (func_in_transaction_11(col_1, NULL::local_table_for_const)));
     -- It should error out since there is non-distributed table dependency exists
     SELECT create_distributed_table('table_to_prop_func_9', 'id');
@@ -730,6 +737,7 @@ HINT:  Distribute "table local_table_for_const" first to distribute "table table
 COMMIT;
 -- Show that function as a part of generated always is supporte
 BEGIN;
+    SET LOCAL citus.create_object_propagation TO deferred;
 	CREATE OR REPLACE FUNCTION non_sense_func_for_generated_always()
 	RETURNS int
 	LANGUAGE plpgsql IMMUTABLE AS
@@ -738,7 +746,6 @@ BEGIN;
 	    return 1;
 	END;
 	$$;
-    -- Functions shouldn't be propagated within transaction
     SELECT pg_identify_object_as_address(classid, objid, objsubid) from pg_catalog.pg_dist_object where objid = 'function_propagation_schema.non_sense_func_for_generated_always'::regproc::oid;
  pg_identify_object_as_address
 ---------------------------------------------------------------------
@@ -764,6 +771,7 @@ BEGIN;
 COMMIT;
 -- Show that functions depending table via rule are also distributed
 BEGIN;
+    SET LOCAL citus.create_object_propagation TO deferred;
     CREATE OR REPLACE FUNCTION func_for_rule()
     RETURNS int
     LANGUAGE plpgsql STABLE AS
@@ -772,7 +780,7 @@ BEGIN;
         return 4;
     END;
     $$;
-    -- Functions shouldn't be propagated within transaction
+    RESET citus.create_object_propagation;
     SELECT pg_identify_object_as_address(classid, objid, objsubid) from pg_catalog.pg_dist_object where objid = 'function_propagation_schema.func_for_rule'::regproc::oid;
  pg_identify_object_as_address
 ---------------------------------------------------------------------
@@ -805,6 +813,7 @@ SELECT * FROM run_command_on_workers($$SELECT pg_identify_object_as_address(clas
 
 -- Show that functions as partitioning functions are supported
 BEGIN;
+    SET LOCAL citus.create_object_propagation TO deferred;
 	CREATE OR REPLACE FUNCTION non_sense_func_for_partitioning(int)
 	RETURNS int
 	LANGUAGE plpgsql IMMUTABLE AS
@@ -813,7 +822,6 @@ BEGIN;
 	    return 1;
 	END;
 	$$;
-    -- Functions shouldn't be propagated within transaction
     SELECT pg_identify_object_as_address(classid, objid, objsubid) from pg_catalog.pg_dist_object where objid = 'function_propagation_schema.non_sense_func_for_partitioning'::regproc::oid;
  pg_identify_object_as_address
 ---------------------------------------------------------------------
@@ -844,6 +852,7 @@ SELECT * FROM run_command_on_workers($$SELECT pg_identify_object_as_address(clas
 
 -- Test function dependency on citus local table
 BEGIN;
+    SET LOCAL citus.create_object_propagation TO deferred;
     CREATE OR REPLACE FUNCTION func_in_transaction_for_local_table()
     RETURNS int
     LANGUAGE plpgsql AS
@@ -852,7 +861,6 @@ BEGIN;
         return 1;
     END;
     $$;
-    -- Function shouldn't be propagated within transaction
     SELECT pg_identify_object_as_address(classid, objid, objsubid) from pg_catalog.pg_dist_object where objid = 'function_propagation_schema.func_in_transaction_for_local_table'::regproc::oid;
  pg_identify_object_as_address
 ---------------------------------------------------------------------
@@ -882,6 +890,7 @@ NOTICE:  localhost:xxxxx is the coordinator and already contains metadata, skipp
 ROLLBACK;
 -- Show that having a function dependency on exlude also works
 BEGIN;
+    SET LOCAL citus.create_object_propagation TO deferred;
     CREATE OR REPLACE FUNCTION exclude_bool_func()
     RETURNS boolean
     LANGUAGE plpgsql IMMUTABLE AS
@@ -890,7 +899,6 @@ BEGIN;
         return true;
     END;
     $$;
-    -- Functions shouldn't be propagated within transaction
     SELECT pg_identify_object_as_address(classid, objid, objsubid) from pg_catalog.pg_dist_object where objid = 'function_propagation_schema.exclude_bool_func'::regproc::oid;
  pg_identify_object_as_address
 ---------------------------------------------------------------------
@@ -921,6 +929,7 @@ SELECT * FROM run_command_on_workers($$SELECT pg_identify_object_as_address(clas
 
 -- Show that having a function dependency for index also works
 BEGIN;
+    SET LOCAL citus.create_object_propagation TO deferred;
     CREATE OR REPLACE FUNCTION func_for_index_predicate(col_1 int)
     RETURNS boolean
     LANGUAGE plpgsql IMMUTABLE AS
@@ -929,7 +938,6 @@ BEGIN;
         return col_1 > 5;
     END;
     $$;
-    -- Functions shouldn't be propagated within transaction
     SELECT pg_identify_object_as_address(classid, objid, objsubid) from pg_catalog.pg_dist_object where objid = 'function_propagation_schema.func_for_index_predicate'::regproc::oid;
  pg_identify_object_as_address
 ---------------------------------------------------------------------
@@ -978,6 +986,9 @@ BEGIN;
         return 5;
     END;
     $$;
+WARNING:  "function func_for_func_dep_2(func_dep_table)" has dependency to "table func_dep_table" that is not in Citus' metadata
+DETAIL:  "function func_for_func_dep_2(func_dep_table)" will be created only locally
+HINT:  Distribute "table func_dep_table" first to distribute "function func_for_func_dep_2(func_dep_table)"
     SELECT create_distributed_table('func_dep_table', 'a');
  create_distributed_table
 ---------------------------------------------------------------------
@@ -1002,6 +1013,7 @@ SELECT * FROM run_command_on_workers($$SELECT pg_identify_object_as_address(clas
 
 -- Test function with SQL language and sequence dependency
 BEGIN;
+    SET LOCAL citus.create_object_propagation TO deferred;
     CREATE OR REPLACE FUNCTION func_in_transaction_def_with_seq(val bigint)
     RETURNS bigint
     LANGUAGE SQL AS
@@ -1014,7 +1026,6 @@ BEGIN;
     $$
     SELECT func_in_transaction_def_with_seq(val);
     $$;
-    -- Function shouldn't be propagated within transaction
     SELECT pg_identify_object_as_address(classid, objid, objsubid) from pg_catalog.pg_dist_object where objid = 'function_propagation_schema.func_in_transaction_def_with_seq'::regproc::oid;
  pg_identify_object_as_address
 ---------------------------------------------------------------------

src/test/regress/expected/metadata_sync_helpers.out

@@ -694,6 +694,7 @@ BEGIN TRANSACTION ISOLATION LEVEL READ COMMITTED;
 
     SET application_name to 'citus_internal gpid=10000000001';
     \set VERBOSITY terse
+	SET citus.enable_ddl_propagation TO OFF;
     CREATE FUNCTION distribution_test_function(int) RETURNS int
     AS $$ SELECT $1 $$
     LANGUAGE SQL;

src/test/regress/expected/shard_move_deferred_delete.out

@@ -233,22 +233,27 @@ SET search_path TO shard_move_deferred_delete;
 SELECT master_move_shard_placement(20000001, 'localhost', :worker_2_port, 'localhost', :worker_1_port);
 ERROR:  not enough empty space on node if the shard is moved, actual available space after move will be 108 bytes, desired available space after move is 850 bytes,estimated size increase on node after move is 8192 bytes.
 HINT:  consider lowering citus.desired_percent_disk_available_after_move.
--- Restore the original function
+-- Restore the original function on workers
-SELECT run_command_on_workers($cmd$
+\c - - - :worker_1_port
-    CREATE OR REPLACE FUNCTION pg_catalog.citus_local_disk_space_stats(
+SET citus.enable_metadata_sync TO OFF;
-    OUT available_disk_size bigint,
+CREATE OR REPLACE FUNCTION pg_catalog.citus_local_disk_space_stats(
-    OUT total_disk_size bigint)
+OUT available_disk_size bigint,
-    RETURNS record
+OUT total_disk_size bigint)
-    LANGUAGE C STRICT
+RETURNS record
-    AS 'citus', $$citus_local_disk_space_stats$$;
+LANGUAGE C STRICT
-    COMMENT ON FUNCTION pg_catalog.citus_local_disk_space_stats()
+AS 'citus', $$citus_local_disk_space_stats$$;
-    IS 'returns statistics on available disk space on the local node';
+COMMENT ON FUNCTION pg_catalog.citus_local_disk_space_stats()
-$cmd$);
+IS 'returns statistics on available disk space on the local node';
-        run_command_on_workers
+\c - - - :worker_2_port
----------------------------------------------------------------------
+SET citus.enable_metadata_sync TO OFF;
- (localhost,57637,t,"CREATE FUNCTION")
+CREATE OR REPLACE FUNCTION pg_catalog.citus_local_disk_space_stats(
- (localhost,57638,t,"CREATE FUNCTION")
+OUT available_disk_size bigint,
-(2 rows)
+OUT total_disk_size bigint)
-
+RETURNS record
+LANGUAGE C STRICT
+AS 'citus', $$citus_local_disk_space_stats$$;
+COMMENT ON FUNCTION pg_catalog.citus_local_disk_space_stats()
+IS 'returns statistics on available disk space on the local node';
+\c - - - :master_port
 DROP SCHEMA shard_move_deferred_delete CASCADE;
-NOTICE:  drop cascades to table t1
+NOTICE:  drop cascades to table shard_move_deferred_delete.t1

src/test/regress/expected/tableam.out

@@ -12,6 +12,7 @@ SET citus.shard_count TO 4;
 create schema test_tableam;
 set search_path to test_tableam;
 SELECT public.run_command_on_coordinator_and_workers($Q$
+  SET citus.enable_ddl_propagation TO off;
 	CREATE FUNCTION fake_am_handler(internal)
 	RETURNS table_am_handler
 	AS 'citus'
@@ -26,8 +27,6 @@ $Q$);
 -- Since Citus assumes access methods are part of the extension, make fake_am
 -- owned manually to be able to pass checks on Citus while distributing tables.
 ALTER EXTENSION citus ADD ACCESS METHOD fake_am;
-NOTICE:  Citus does not propagate adding/dropping member objects
-HINT:  You can add/drop the member objects on the workers as well.
 --
 -- Hash distributed table using a non-default table access method
 --

src/test/regress/spec/isolation_blocking_move_multi_shard_commands_on_mx.spec

@@ -3,6 +3,7 @@
 
 setup
 {
+  SET citus.enable_metadata_sync TO off;
   CREATE OR REPLACE FUNCTION start_session_level_connection_to_node(text, integer)
       RETURNS void
       LANGUAGE C STRICT VOLATILE
@@ -17,6 +18,7 @@ setup
       RETURNS void
       LANGUAGE C STRICT VOLATILE
       AS 'citus', $$stop_session_level_connection_to_node$$;
+  RESET citus.enable_metadata_sync;
 
   SELECT citus_internal.replace_isolation_tester_func();
   SELECT citus_internal.refresh_isolation_tester_prepared_statement();

src/test/regress/spec/isolation_blocking_move_single_shard_commands_on_mx.spec

@@ -2,6 +2,7 @@
 // so setting the corresponding shard here is useful
 setup
 {
+  SET citus.enable_ddl_propagation TO OFF;
   CREATE OR REPLACE FUNCTION start_session_level_connection_to_node(text, integer)
       RETURNS void
       LANGUAGE C STRICT VOLATILE
@@ -16,6 +17,7 @@ setup
       RETURNS void
       LANGUAGE C STRICT VOLATILE
       AS 'citus', $$stop_session_level_connection_to_node$$;
+  RESET citus.enable_ddl_propagation;
 
   SELECT citus_internal.replace_isolation_tester_func();
   SELECT citus_internal.refresh_isolation_tester_prepared_statement();

src/test/regress/spec/isolation_dump_global_wait_edges.spec

@@ -6,12 +6,14 @@ setup
     SELECT create_distributed_table('distributed_table', 'x');
     INSERT INTO distributed_table VALUES (1,0);
 
+    SET citus.enable_ddl_propagation TO OFF;
     CREATE OR REPLACE FUNCTION get_adjacency_list_wait_graph(OUT transactionNumber int, OUT waitingTransactionNumbers cstring)
     RETURNS SETOF RECORD
     LANGUAGE C STRICT
     AS 'citus', $$get_adjacency_list_wait_graph$$;
     COMMENT ON FUNCTION get_adjacency_list_wait_graph(OUT transactionNumber int, OUT waitingTransactionNumbers cstring)
     IS 'returns flattened wait graph';
+    RESET citus.enable_ddl_propagation;
 }
 
 teardown

src/test/regress/spec/isolation_max_client_connections.spec

@@ -11,10 +11,12 @@ setup
 	GRANT USAGE ON SCHEMA public TO my_user;
 	GRANT SELECT ON TABLE my_table TO my_user;
 
+	SET citus.enable_ddl_propagation TO OFF;
 	CREATE FUNCTION make_external_connection_to_node(text,int,text,text)
 	RETURNS void
 	AS 'citus'
 	LANGUAGE C STRICT;
+	RESET citus.enable_ddl_propagation;
 
 	SELECT run_command_on_workers('ALTER SYSTEM SET citus.max_client_connections TO 1');
 	SELECT run_command_on_workers('SELECT pg_reload_conf()');

src/test/regress/spec/isolation_mx_common.include.spec

@@ -2,6 +2,7 @@
 // ready for testing MX functionalities.
 setup
 {
+    SET citus.enable_metadata_sync TO off;
     CREATE OR REPLACE FUNCTION start_session_level_connection_to_node(text, integer)
         RETURNS void
         LANGUAGE C STRICT VOLATILE
@@ -26,6 +27,7 @@ setup
         RETURNS void
         LANGUAGE C STRICT VOLATILE
         AS 'citus', $$stop_session_level_connection_to_node$$;
+    RESET citus.enable_metadata_sync;
 
     SELECT citus_internal.replace_isolation_tester_func();
     SELECT citus_internal.refresh_isolation_tester_prepared_statement();

src/test/regress/spec/isolation_rebalancer_deferred_drop.spec

@@ -2,7 +2,7 @@
 // so setting the corresponding shard here is useful
 setup
 {
-
+    SET citus.enable_metadata_sync TO off;
     CREATE OR REPLACE FUNCTION run_try_drop_marked_shards()
     RETURNS VOID
     AS 'citus'
@@ -31,6 +31,7 @@ CREATE OR REPLACE PROCEDURE isolation_cleanup_orphaned_shards()
     AS 'citus', $$isolation_cleanup_orphaned_shards$$;
 COMMENT ON PROCEDURE isolation_cleanup_orphaned_shards()
     IS 'cleanup orphaned shards';
+    RESET citus.enable_metadata_sync;
 
     SET citus.next_shard_id to 120000;
 	SET citus.shard_count TO 8;

src/test/regress/sql/citus_local_table_triggers.sql

@@ -27,7 +27,6 @@ ALTER TABLE citus_local_table ADD CONSTRAINT fkey_to_dummy_1 FOREIGN KEY (value)
 
 BEGIN;
     CREATE TABLE distributed_table(value int);
-    SELECT create_distributed_table('distributed_table', 'value');
     CREATE FUNCTION insert_42() RETURNS trigger AS $insert_42$
     BEGIN
         INSERT INTO distributed_table VALUES (42);
@@ -35,6 +34,8 @@ BEGIN;
     END;
     $insert_42$ LANGUAGE plpgsql;
 
+    SELECT create_distributed_table('distributed_table', 'value');
+
     CREATE TRIGGER insert_42_trigger
     AFTER DELETE ON citus_local_table
     FOR EACH ROW EXECUTE FUNCTION insert_42();

src/test/regress/sql/function_propagation.sql

@@ -192,6 +192,7 @@ $$;
 
 -- Function as a default column
 BEGIN;
+    SET LOCAL citus.create_object_propagation TO deferred;
     CREATE OR REPLACE FUNCTION func_in_transaction_def()
     RETURNS int
     LANGUAGE plpgsql AS
@@ -201,7 +202,6 @@ BEGIN;
     END;
     $$;
 
-    -- Function shouldn't be propagated within transaction
     SELECT pg_identify_object_as_address(classid, objid, objsubid) from pg_catalog.pg_dist_object where objid = 'function_propagation_schema.func_in_transaction_def'::regproc::oid;
 
     CREATE TABLE table_to_prop_func(id int, col_1 int default func_in_transaction_def());
@@ -217,6 +217,7 @@ SELECT * FROM run_command_on_workers($$SELECT pg_identify_object_as_address(clas
 
 -- Multiple functions as a default column
 BEGIN;
+    SET LOCAL citus.create_object_propagation TO deferred;
     CREATE OR REPLACE FUNCTION func_in_transaction_1()
     RETURNS int
     LANGUAGE plpgsql AS
@@ -235,7 +236,6 @@ BEGIN;
     END;
     $$;
 
-    -- Functions shouldn't be propagated within transaction
     SELECT pg_identify_object_as_address(classid, objid, objsubid) from pg_catalog.pg_dist_object where objid = 'function_propagation_schema.func_in_transaction_1'::regproc::oid;
     SELECT pg_identify_object_as_address(classid, objid, objsubid) from pg_catalog.pg_dist_object where objid = 'function_propagation_schema.func_in_transaction_2'::regproc::oid;
 
@@ -275,8 +275,8 @@ COMMIT;
 -- Adding a column with default value should propagate the function
 BEGIN;
     CREATE TABLE table_to_prop_func_4(id int);
-    SELECT create_distributed_table('table_to_prop_func_4', 'id');
 
+    SET LOCAL citus.create_object_propagation TO deferred;
     CREATE OR REPLACE FUNCTION func_in_transaction_4()
     RETURNS int
     LANGUAGE plpgsql AS
@@ -286,7 +286,8 @@ BEGIN;
     END;
     $$;
 
-    -- Function shouldn't be propagated within transaction
+    SELECT create_distributed_table('table_to_prop_func_4', 'id');
+
     SELECT pg_identify_object_as_address(classid, objid, objsubid) from pg_catalog.pg_dist_object where objid = 'function_propagation_schema.func_in_transaction_4'::regproc::oid;
 
     ALTER TABLE table_to_prop_func_4 ADD COLUMN col_1 int default function_propagation_schema.func_in_transaction_4();
@@ -303,6 +304,7 @@ SELECT * FROM run_command_on_workers($$SELECT pg_identify_object_as_address(clas
 BEGIN;
     CREATE TABLE non_dist_table_for_function(id int);
 
+    SET LOCAL citus.create_object_propagation TO deferred;
     CREATE OR REPLACE FUNCTION non_dist_func(col_1 non_dist_table_for_function)
     RETURNS int
     LANGUAGE plpgsql AS
@@ -322,6 +324,7 @@ ROLLBACK;
 
 -- Adding multiple columns with default values should propagate the function
 BEGIN;
+    SET LOCAL citus.create_object_propagation TO deferred;
     CREATE OR REPLACE FUNCTION func_in_transaction_5()
     RETURNS int
     LANGUAGE plpgsql AS
@@ -340,8 +343,6 @@ BEGIN;
     END;
     $$;
 
-
-    -- Functions shouldn't be propagated within transaction
     SELECT pg_identify_object_as_address(classid, objid, objsubid) from pg_catalog.pg_dist_object where objid = 'function_propagation_schema.func_in_transaction_5'::regproc::oid;
     SELECT pg_identify_object_as_address(classid, objid, objsubid) from pg_catalog.pg_dist_object where objid = 'function_propagation_schema.func_in_transaction_6'::regproc::oid;
 
@@ -359,6 +360,7 @@ SELECT * FROM run_command_on_workers($$SELECT pg_identify_object_as_address(clas
 
 -- Adding a constraint with function check should propagate the function
 BEGIN;
+    SET LOCAL citus.create_object_propagation TO deferred;
     CREATE OR REPLACE FUNCTION func_in_transaction_7(param_1 int)
     RETURNS boolean
     LANGUAGE plpgsql AS
@@ -368,7 +370,6 @@ BEGIN;
     END;
     $$;
 
-    -- Functions shouldn't be propagated within transaction
     SELECT pg_identify_object_as_address(classid, objid, objsubid) from pg_catalog.pg_dist_object where objid = 'function_propagation_schema.func_in_transaction_7'::regproc::oid;
 
     CREATE TABLE table_to_prop_func_6(id int, col_1 int check (function_propagation_schema.func_in_transaction_7(col_1)));
@@ -384,6 +385,7 @@ SELECT * FROM run_command_on_workers($$SELECT pg_identify_object_as_address(clas
 
 -- Adding a constraint with multiple functions check should propagate the function
 BEGIN;
+    SET LOCAL citus.create_object_propagation TO deferred;
     CREATE OR REPLACE FUNCTION func_in_transaction_8(param_1 int)
     RETURNS boolean
     LANGUAGE plpgsql AS
@@ -402,7 +404,6 @@ BEGIN;
     END;
     $$;
 
-    -- Functions shouldn't be propagated within transaction
     SELECT pg_identify_object_as_address(classid, objid, objsubid) from pg_catalog.pg_dist_object where objid = 'function_propagation_schema.func_in_transaction_8'::regproc::oid;
     SELECT pg_identify_object_as_address(classid, objid, objsubid) from pg_catalog.pg_dist_object where objid = 'function_propagation_schema.func_in_transaction_9'::regproc::oid;
 
@@ -424,6 +425,7 @@ BEGIN;
     CREATE TABLE table_to_prop_func_8(id int, col_1 int);
     SELECT create_distributed_table('table_to_prop_func_8', 'id');
 
+    SET LOCAL citus.create_object_propagation TO deferred;
     CREATE OR REPLACE FUNCTION func_in_transaction_10(param_1 int)
     RETURNS boolean
     LANGUAGE plpgsql AS
@@ -433,7 +435,6 @@ BEGIN;
     END;
     $$;
 
-    -- Functions shouldn't be propagated within transaction
     SELECT pg_identify_object_as_address(classid, objid, objsubid) from pg_catalog.pg_dist_object where objid = 'function_propagation_schema.func_in_transaction_10'::regproc::oid;
 
     ALTER TABLE table_to_prop_func_8 ADD CONSTRAINT col1_check CHECK (function_propagation_schema.func_in_transaction_10(col_1));
@@ -468,7 +469,7 @@ COMMIT;
 
 -- Show that function as a part of generated always is supporte
 BEGIN;
-
+    SET LOCAL citus.create_object_propagation TO deferred;
 	CREATE OR REPLACE FUNCTION non_sense_func_for_generated_always()
 	RETURNS int
 	LANGUAGE plpgsql IMMUTABLE AS
@@ -478,7 +479,6 @@ BEGIN;
 	END;
 	$$;
 
-    -- Functions shouldn't be propagated within transaction
     SELECT pg_identify_object_as_address(classid, objid, objsubid) from pg_catalog.pg_dist_object where objid = 'function_propagation_schema.non_sense_func_for_generated_always'::regproc::oid;
 
 	CREATE TABLE people (
@@ -495,6 +495,7 @@ COMMIT;
 
 -- Show that functions depending table via rule are also distributed
 BEGIN;
+    SET LOCAL citus.create_object_propagation TO deferred;
     CREATE OR REPLACE FUNCTION func_for_rule()
     RETURNS int
     LANGUAGE plpgsql STABLE AS
@@ -503,8 +504,8 @@ BEGIN;
         return 4;
     END;
     $$;
+    RESET citus.create_object_propagation;
 
-    -- Functions shouldn't be propagated within transaction
     SELECT pg_identify_object_as_address(classid, objid, objsubid) from pg_catalog.pg_dist_object where objid = 'function_propagation_schema.func_for_rule'::regproc::oid;
 
     CREATE TABLE table_1_for_rule(id int, col_1 int);
@@ -524,7 +525,7 @@ SELECT * FROM run_command_on_workers($$SELECT pg_identify_object_as_address(clas
 
 -- Show that functions as partitioning functions are supported
 BEGIN;
-
+    SET LOCAL citus.create_object_propagation TO deferred;
 	CREATE OR REPLACE FUNCTION non_sense_func_for_partitioning(int)
 	RETURNS int
 	LANGUAGE plpgsql IMMUTABLE AS
@@ -534,7 +535,6 @@ BEGIN;
 	END;
 	$$;
 
-    -- Functions shouldn't be propagated within transaction
     SELECT pg_identify_object_as_address(classid, objid, objsubid) from pg_catalog.pg_dist_object where objid = 'function_propagation_schema.non_sense_func_for_partitioning'::regproc::oid;
 
     CREATE TABLE partitioned_table_to_test_func_prop(id INT, a INT) PARTITION BY RANGE (non_sense_func_for_partitioning(id));
@@ -551,6 +551,7 @@ SELECT * FROM run_command_on_workers($$SELECT pg_identify_object_as_address(clas
 
 -- Test function dependency on citus local table
 BEGIN;
+    SET LOCAL citus.create_object_propagation TO deferred;
     CREATE OR REPLACE FUNCTION func_in_transaction_for_local_table()
     RETURNS int
     LANGUAGE plpgsql AS
@@ -560,7 +561,6 @@ BEGIN;
     END;
     $$;
 
-    -- Function shouldn't be propagated within transaction
     SELECT pg_identify_object_as_address(classid, objid, objsubid) from pg_catalog.pg_dist_object where objid = 'function_propagation_schema.func_in_transaction_for_local_table'::regproc::oid;
 
     CREATE TABLE citus_local_table_to_test_func(l1 int DEFAULT func_in_transaction_for_local_table());
@@ -573,6 +573,7 @@ ROLLBACK;
 
 -- Show that having a function dependency on exlude also works
 BEGIN;
+    SET LOCAL citus.create_object_propagation TO deferred;
     CREATE OR REPLACE FUNCTION exclude_bool_func()
     RETURNS boolean
     LANGUAGE plpgsql IMMUTABLE AS
@@ -582,7 +583,6 @@ BEGIN;
     END;
     $$;
 
-    -- Functions shouldn't be propagated within transaction
     SELECT pg_identify_object_as_address(classid, objid, objsubid) from pg_catalog.pg_dist_object where objid = 'function_propagation_schema.exclude_bool_func'::regproc::oid;
 
     CREATE TABLE exclusion_func_prop_table (id int, EXCLUDE USING btree (id WITH =) WHERE (exclude_bool_func()));
@@ -598,6 +598,7 @@ SELECT * FROM run_command_on_workers($$SELECT pg_identify_object_as_address(clas
 
 -- Show that having a function dependency for index also works
 BEGIN;
+    SET LOCAL citus.create_object_propagation TO deferred;
     CREATE OR REPLACE FUNCTION func_for_index_predicate(col_1 int)
     RETURNS boolean
     LANGUAGE plpgsql IMMUTABLE AS
@@ -607,7 +608,6 @@ BEGIN;
     END;
     $$;
 
-    -- Functions shouldn't be propagated within transaction
     SELECT pg_identify_object_as_address(classid, objid, objsubid) from pg_catalog.pg_dist_object where objid = 'function_propagation_schema.func_for_index_predicate'::regproc::oid;
 
     CREATE TABLE table_to_check_func_index_dep (id int, col_2 int);
@@ -657,6 +657,7 @@ SELECT * FROM run_command_on_workers($$SELECT pg_identify_object_as_address(clas
 
 -- Test function with SQL language and sequence dependency
 BEGIN;
+    SET LOCAL citus.create_object_propagation TO deferred;
     CREATE OR REPLACE FUNCTION func_in_transaction_def_with_seq(val bigint)
     RETURNS bigint
     LANGUAGE SQL AS
@@ -671,7 +672,6 @@ BEGIN;
     SELECT func_in_transaction_def_with_seq(val);
     $$;
 
-    -- Function shouldn't be propagated within transaction
     SELECT pg_identify_object_as_address(classid, objid, objsubid) from pg_catalog.pg_dist_object where objid = 'function_propagation_schema.func_in_transaction_def_with_seq'::regproc::oid;
 
     CREATE SEQUENCE myseq;

src/test/regress/sql/metadata_sync_helpers.sql

@@ -424,6 +424,7 @@ BEGIN TRANSACTION ISOLATION LEVEL READ COMMITTED;
     SET application_name to 'citus_internal gpid=10000000001';
     \set VERBOSITY terse
 
+	SET citus.enable_ddl_propagation TO OFF;
     CREATE FUNCTION distribution_test_function(int) RETURNS int
     AS $$ SELECT $1 $$
     LANGUAGE SQL;

src/test/regress/sql/shard_move_deferred_delete.sql

@@ -167,17 +167,30 @@ SET search_path TO shard_move_deferred_delete;
 -- When there would not be enough free space left after the move, the move should fail
 SELECT master_move_shard_placement(20000001, 'localhost', :worker_2_port, 'localhost', :worker_1_port);
 
--- Restore the original function
+-- Restore the original function on workers
-SELECT run_command_on_workers($cmd$
+\c - - - :worker_1_port
-    CREATE OR REPLACE FUNCTION pg_catalog.citus_local_disk_space_stats(
+SET citus.enable_metadata_sync TO OFF;
-    OUT available_disk_size bigint,
-    OUT total_disk_size bigint)
-    RETURNS record
-    LANGUAGE C STRICT
-    AS 'citus', $$citus_local_disk_space_stats$$;
-    COMMENT ON FUNCTION pg_catalog.citus_local_disk_space_stats()
-    IS 'returns statistics on available disk space on the local node';
-$cmd$);
 
+CREATE OR REPLACE FUNCTION pg_catalog.citus_local_disk_space_stats(
+OUT available_disk_size bigint,
+OUT total_disk_size bigint)
+RETURNS record
+LANGUAGE C STRICT
+AS 'citus', $$citus_local_disk_space_stats$$;
+COMMENT ON FUNCTION pg_catalog.citus_local_disk_space_stats()
+IS 'returns statistics on available disk space on the local node';
 
+\c - - - :worker_2_port
+SET citus.enable_metadata_sync TO OFF;
+
+CREATE OR REPLACE FUNCTION pg_catalog.citus_local_disk_space_stats(
+OUT available_disk_size bigint,
+OUT total_disk_size bigint)
+RETURNS record
+LANGUAGE C STRICT
+AS 'citus', $$citus_local_disk_space_stats$$;
+COMMENT ON FUNCTION pg_catalog.citus_local_disk_space_stats()
+IS 'returns statistics on available disk space on the local node';
+
+\c - - - :master_port
 DROP SCHEMA shard_move_deferred_delete CASCADE;

src/test/regress/sql/tableam.sql

@@ -15,6 +15,7 @@ create schema test_tableam;
 set search_path to test_tableam;
 
 SELECT public.run_command_on_coordinator_and_workers($Q$
+	SET citus.enable_ddl_propagation TO off;
 	CREATE FUNCTION fake_am_handler(internal)
 	RETURNS table_am_handler
 	AS 'citus'