Use coordinated transaction when there are multiple queries per task

src/backend/distributed/executor/adaptive_executor.c

@@ -1436,6 +1436,15 @@ DistributedExecutionRequiresRollback(List *taskList)
 		return true;
 	}
 
+	if (task->queryCount > 1)
+	{
+		/*
+		 * When there are multiple sequential queries in a task
+		 * we need to run those as a transaction.
+		 */
+		return true;
+	}
+
 	return false;
 }
 

src/test/regress/expected/multi_fix_partition_shard_index_names.out

@@ -644,6 +644,8 @@ ALTER INDEX p1_dist_col_idx RENAME TO p1_dist_col_idx_renamed;
 SET citus.log_remote_commands TO ON;
 CREATE TABLE p2(dist_col int NOT NULL, another_col int, partition_col timestamp NOT NULL, name text) USING columnar;
 ALTER TABLE parent_table ATTACH PARTITION p2 FOR VALUES FROM ('2019-01-01') TO ('2020-01-01');
+NOTICE:  issuing BEGIN TRANSACTION ISOLATION LEVEL READ COMMITTED;SELECT assign_distributed_transaction_id(xx, xx, 'xxxxxxx');
+DETAIL:  on server postgres@localhost:xxxxx connectionId: xxxxxxx
 NOTICE:  issuing SELECT worker_apply_shard_ddl_command (915002, 'fix_idx_names', 'CREATE TABLE fix_idx_names.p2 (dist_col integer NOT NULL, another_col integer, partition_col timestamp without time zone NOT NULL, name text) USING columnar')
 DETAIL:  on server postgres@localhost:xxxxx connectionId: xxxxxxx
 NOTICE:  issuing SELECT alter_columnar_table_set('fix_idx_names.p2_915002', chunk_group_row_limit => 10000, stripe_row_limit => 150000, compression_level => 3, compression => 'zstd');
@@ -652,8 +654,6 @@ NOTICE:  issuing SELECT worker_apply_shard_ddl_command (915002, 'fix_idx_names',
 DETAIL:  on server postgres@localhost:xxxxx connectionId: xxxxxxx
 NOTICE:  issuing BEGIN TRANSACTION ISOLATION LEVEL READ COMMITTED;SELECT assign_distributed_transaction_id(xx, xx, 'xxxxxxx');
 DETAIL:  on server postgres@localhost:xxxxx connectionId: xxxxxxx
-NOTICE:  issuing BEGIN TRANSACTION ISOLATION LEVEL READ COMMITTED;SELECT assign_distributed_transaction_id(xx, xx, 'xxxxxxx');
-DETAIL:  on server postgres@localhost:xxxxx connectionId: xxxxxxx
 NOTICE:  issuing SET citus.enable_ddl_propagation TO 'off'
 DETAIL:  on server postgres@localhost:xxxxx connectionId: xxxxxxx
 NOTICE:  issuing SET citus.enable_ddl_propagation TO 'off'

src/test/regress/expected/multi_replicate_reference_table.out

@@ -281,6 +281,20 @@ WHERE colocationid IN
 (1 row)
 
 DROP TABLE replicate_reference_table_rollback;
+-- confirm that there is just 1 node
+SELECT count(*) FROM pg_dist_node;
+ count
+---------------------------------------------------------------------
+     1
+(1 row)
+
+-- test whether we can create distributed objects on a single worker node
+CREATE TABLE cp_test (a int, b text);
+CREATE PROCEDURE ptest1(x text)
+LANGUAGE SQL
+AS $$
+ INSERT INTO cp_test VALUES (1, x);
+$$;
 -- test replicating a reference table when a new node added in TRANSACTION + COMMIT
 CREATE TABLE replicate_reference_table_commit(column1 int);
 SELECT create_reference_table('replicate_reference_table_commit');

src/test/regress/sql/multi_replicate_reference_table.sql

@@ -67,7 +67,6 @@ WHERE
 
 DROP TABLE replicate_reference_table_unhealthy;
 
-
 -- test replicating a reference table when a new node added
 CREATE TABLE replicate_reference_table_valid(column1 int);
 SELECT create_reference_table('replicate_reference_table_valid');
@@ -184,6 +183,15 @@ WHERE colocationid IN
 
 DROP TABLE replicate_reference_table_rollback;
 
+-- confirm that there is just 1 node
+SELECT count(*) FROM pg_dist_node;
+-- test whether we can create distributed objects on a single worker node
+CREATE TABLE cp_test (a int, b text);
+CREATE PROCEDURE ptest1(x text)
+LANGUAGE SQL
+AS $$
+ INSERT INTO cp_test VALUES (1, x);
+$$;
 
 -- test replicating a reference table when a new node added in TRANSACTION + COMMIT
 CREATE TABLE replicate_reference_table_commit(column1 int);