Adds db rename and refreshcoll

src/backend/distributed/commands/database.c

@@ -254,7 +254,7 @@ FilterDistributedDatabases(List *databases)
  * IsSetTablespaceStatement returns true if the statement is a SET TABLESPACE statement,
  * false otherwise.
  */
-static bool
+bool
 IsSetTablespaceStatement(AlterDatabaseStmt *stmt)
 {
 	DefElem *def = NULL;
@@ -293,7 +293,11 @@ PreprocessAlterDatabaseStmt(Node *node, const char *queryString,
 		return NIL;
 	}
 
-	EnsureCoordinator();
+	if (!IsSetTablespaceStatement(stmt))
+	{
+		EnsureCoordinator();
+	}
+
 	SerializeDistributedDDLsOnObjectClassObject(OCLASS_DATABASE, stmt->dbname);
 
 	char *sql = DeparseTreeNode((Node *) stmt);

src/backend/distributed/commands/non_main_db_distribute_object_ops.c

@@ -61,7 +61,7 @@
  */
 typedef struct NonMainDbDistributeObjectOps
 {
-	bool cannotBeExecutedInTransaction;
+	bool (*cannotBeExecutedInTransaction)(Node *parsetree);
 	bool (*checkSupportedObjectType)(Node *parsetree);
 } NonMainDbDistributeObjectOps;
 
@@ -73,41 +73,69 @@ static bool CreateDbStmtCheckSupportedObjectType(Node *node);
 static bool DropDbStmtCheckSupportedObjectType(Node *node);
 static bool GrantStmtCheckSupportedObjectType(Node *node);
 static bool SecLabelStmtCheckSupportedObjectType(Node *node);
+static bool AlterDbStmtCheckSupportedObjectType(Node *node);
+static bool AlterDbCanNotBeExecutedInTransaction(Node *node);
+static bool CanNotBeExecutedInTransaction_True(Node *node);
+static bool CanNotBeExecutedInTransaction_False(Node *node);
+static bool AlterDbRenameCheckSupportedObjectType(Node *node);
+static bool AlterDbOwnerCheckSupportedObjectType(Node *node);
 
 /*
  * OperationArray that holds NonMainDbDistributeObjectOps for different command types.
  */
 static const NonMainDbDistributeObjectOps *const OperationArray[] = {
 	[T_CreateRoleStmt] = &(NonMainDbDistributeObjectOps) {
-		.cannotBeExecutedInTransaction = false,
+		.cannotBeExecutedInTransaction = CanNotBeExecutedInTransaction_False,
 		.checkSupportedObjectType = NULL
 	},
 	[T_DropRoleStmt] = &(NonMainDbDistributeObjectOps) {
-		.cannotBeExecutedInTransaction = false,
+		.cannotBeExecutedInTransaction = CanNotBeExecutedInTransaction_False,
 		.checkSupportedObjectType = NULL
 	},
 	[T_AlterRoleStmt] = &(NonMainDbDistributeObjectOps) {
-		.cannotBeExecutedInTransaction = false,
+		.cannotBeExecutedInTransaction = CanNotBeExecutedInTransaction_False,
 		.checkSupportedObjectType = NULL
 	},
 	[T_GrantRoleStmt] = &(NonMainDbDistributeObjectOps) {
-		.cannotBeExecutedInTransaction = false,
+		.cannotBeExecutedInTransaction = CanNotBeExecutedInTransaction_False,
 		.checkSupportedObjectType = NULL
 	},
 	[T_CreatedbStmt] = &(NonMainDbDistributeObjectOps) {
-		.cannotBeExecutedInTransaction = true,
+		.cannotBeExecutedInTransaction = CanNotBeExecutedInTransaction_True,
 		.checkSupportedObjectType = CreateDbStmtCheckSupportedObjectType
 	},
 	[T_DropdbStmt] = &(NonMainDbDistributeObjectOps) {
-		.cannotBeExecutedInTransaction = true,
+		.cannotBeExecutedInTransaction = CanNotBeExecutedInTransaction_True,
 		.checkSupportedObjectType = DropDbStmtCheckSupportedObjectType
 	},
+	[T_AlterDatabaseSetStmt] = &(NonMainDbDistributeObjectOps) {
+		.cannotBeExecutedInTransaction = CanNotBeExecutedInTransaction_False,
+		.checkSupportedObjectType = NULL
+	},
+	[T_AlterDatabaseStmt] = &(NonMainDbDistributeObjectOps) {
+		.cannotBeExecutedInTransaction = AlterDbCanNotBeExecutedInTransaction,
+		.checkSupportedObjectType = AlterDbStmtCheckSupportedObjectType
+	},
+#if PG_VERSION_NUM >= PG_VERSION_15
+	[T_AlterDatabaseRefreshCollStmt] = &(NonMainDbDistributeObjectOps) {
+		.cannotBeExecutedInTransaction = CanNotBeExecutedInTransaction_False,
+		.checkSupportedObjectType = NULL
+	},
+#endif
+	[T_RenameStmt] = &(NonMainDbDistributeObjectOps) {
+		.cannotBeExecutedInTransaction = CanNotBeExecutedInTransaction_False,
+		.checkSupportedObjectType = AlterDbRenameCheckSupportedObjectType
+	},
+	[T_AlterOwnerStmt] = &(NonMainDbDistributeObjectOps) {
+		.cannotBeExecutedInTransaction = CanNotBeExecutedInTransaction_False,
+		.checkSupportedObjectType = AlterDbOwnerCheckSupportedObjectType
+	},
 	[T_GrantStmt] = &(NonMainDbDistributeObjectOps) {
-		.cannotBeExecutedInTransaction = false,
+		.cannotBeExecutedInTransaction = CanNotBeExecutedInTransaction_False,
 		.checkSupportedObjectType = GrantStmtCheckSupportedObjectType
 	},
 	[T_SecLabelStmt] = &(NonMainDbDistributeObjectOps) {
-		.cannotBeExecutedInTransaction = false,
+		.cannotBeExecutedInTransaction = CanNotBeExecutedInTransaction_False,
 		.checkSupportedObjectType = SecLabelStmtCheckSupportedObjectType
 	},
 };
@@ -132,6 +160,15 @@ static void UnmarkObjectDistributedOnLocalMainDb(uint16 catalogRelId, Oid object
 bool
 RunPreprocessNonMainDBCommand(Node *parsetree)
 {
+	/* NodeTag tag = nodeTag(parsetree); */
+	/* if ( tag == T_AlterDatabaseSetStmt){ */
+	/* 	AlterDatabaseSetStmt *stmt = castNode(AlterDatabaseSetStmt, parsetree); */
+	/* 	if (strcmp(stmt->dbname, MainDb) == 0){ */
+	/* 		return false; */
+	/* 	} */
+	/* } */
+
+
 	if (IsMainDB)
 	{
 		return false;
@@ -150,7 +187,7 @@ RunPreprocessNonMainDBCommand(Node *parsetree)
 	 * transactional visibility issues. We directly route them to main database
 	 * so that we only have to consider one code-path for such commands.
 	 */
-	if (ops->cannotBeExecutedInTransaction)
+	if (ops->cannotBeExecutedInTransaction(parsetree))
 	{
 		IsMainDBCommandInXact = false;
 		RunCitusMainDBQuery((char *) queryString);
@@ -335,6 +372,50 @@ DropDbStmtCheckSupportedObjectType(Node *node)
 }
 
 
+static bool
+AlterDbStmtCheckSupportedObjectType(Node *node)
+{
+	/*
+	 * We don't try to send the query to the main database if the ALTER
+	 * DATABASE command is for the main database itself, this is a very
+	 * rare case but it's exercised by our test suite.
+	 */
+	AlterDatabaseStmt *stmt = castNode(AlterDatabaseStmt, node);
+	if (!IsSetTablespaceStatement(stmt))
+	{
+		return true;
+	}
+	else
+	{
+		return IsSetTablespaceStatement(stmt) && strcmp(stmt->dbname, MainDb) != 0;
+	}
+}
+
+
+static bool
+AlterDbCanNotBeExecutedInTransaction(Node *node)
+{
+	AlterDatabaseStmt *stmt = castNode(AlterDatabaseStmt, node);
+	return IsSetTablespaceStatement(stmt);
+}
+
+
+static bool
+AlterDbRenameCheckSupportedObjectType(Node *node)
+{
+	RenameStmt *stmt = castNode(RenameStmt, node);
+	return stmt->renameType == OBJECT_DATABASE;
+}
+
+
+static bool
+AlterDbOwnerCheckSupportedObjectType(Node *node)
+{
+	AlterOwnerStmt *stmt = castNode(AlterOwnerStmt, node);
+	return stmt->objectType == OBJECT_DATABASE;
+}
+
+
 static bool
 GrantStmtCheckSupportedObjectType(Node *node)
 {
@@ -349,3 +430,17 @@ SecLabelStmtCheckSupportedObjectType(Node *node)
 	SecLabelStmt *stmt = castNode(SecLabelStmt, node);
 	return stmt->objtype == OBJECT_ROLE;
 }
+
+
+static bool
+CanNotBeExecutedInTransaction_True(Node *node)
+{
+	return true;
+}
+
+
+static bool
+CanNotBeExecutedInTransaction_False(Node *node)
+{
+	return false;
+}

src/backend/distributed/commands/utility_hook.c

@@ -252,8 +252,12 @@ citus_ProcessUtility(PlannedStmt *pstmt,
 		 * and if the command is a node-wide object management command that we
 		 * support from non-main databases.
 		 */
+		bool shouldSkipPrevUtilityHook = false;
 
-		bool shouldSkipPrevUtilityHook = RunPreprocessNonMainDBCommand(parsetree);
+		if (EnableDDLPropagation)
+		{
+			shouldSkipPrevUtilityHook = RunPreprocessNonMainDBCommand(parsetree);
+		}
 
 		if (!shouldSkipPrevUtilityHook)
 		{
@@ -264,8 +268,10 @@ citus_ProcessUtility(PlannedStmt *pstmt,
 			PrevProcessUtility(pstmt, queryString, false, context,
 							   params, queryEnv, dest, completionTag);
 		}
-
+		if (EnableDDLPropagation)
-		RunPostprocessNonMainDBCommand(parsetree);
+		{
+			RunPostprocessNonMainDBCommand(parsetree);
+		}
 
 		return;
 	}

src/include/distributed/commands.h

@@ -256,6 +256,7 @@ extern List * PreprocessAlterDatabaseRenameStmt(Node *node, const char *queryStr
 extern List * PostprocessAlterDatabaseRenameStmt(Node *node, const char *queryString);
 extern void EnsureSupportedCreateDatabaseCommand(CreatedbStmt *stmt);
 extern char * CreateDatabaseDDLCommand(Oid dbId);
+extern bool IsSetTablespaceStatement(AlterDatabaseStmt *stmt);
 
 
 /* domain.c - forward declarations */

src/test/regress/expected/minimal_cluster_management.out

@@ -1,3 +1,44 @@
+ALTER SYSTEM SET citus.enable_ddl_propagation = 'true';
+SELECT pg_reload_conf();
+ pg_reload_conf
+---------------------------------------------------------------------
+ t
+(1 row)
+
+\c - - - :worker_1_port
+ALTER SYSTEM SET citus.enable_ddl_propagation = 'true';
+SELECT pg_reload_conf();
+ pg_reload_conf
+---------------------------------------------------------------------
+ t
+(1 row)
+
+\c - - - :worker_2_port
+ALTER SYSTEM SET citus.enable_ddl_propagation = 'true';
+SELECT pg_reload_conf();
+ pg_reload_conf
+---------------------------------------------------------------------
+ t
+(1 row)
+
+\c - - - :master_port
+\c - - - :worker_1_port
+ALTER SYSTEM SET citus.enable_ddl_propagation = 'true';
+SELECT pg_reload_conf();
+ pg_reload_conf
+---------------------------------------------------------------------
+ t
+(1 row)
+
+\c - - - :worker_2_port
+ALTER SYSTEM SET citus.enable_ddl_propagation = 'true';
+SELECT pg_reload_conf();
+ pg_reload_conf
+---------------------------------------------------------------------
+ t
+(1 row)
+
+\c - - - :master_port
 SET citus.next_shard_id TO 1220000;
 ALTER SEQUENCE pg_catalog.pg_dist_colocationid_seq RESTART 1390000;
 ALTER SEQUENCE pg_catalog.pg_dist_groupid_seq RESTART 1;

src/test/regress/expected/multi_cluster_management.out

@@ -1,3 +1,27 @@
+ALTER SYSTEM SET citus.enable_ddl_propagation = 'true';
+SELECT pg_reload_conf();
+ pg_reload_conf
+---------------------------------------------------------------------
+ t
+(1 row)
+
+\c - - - :worker_1_port
+ALTER SYSTEM SET citus.enable_ddl_propagation = 'true';
+SELECT pg_reload_conf();
+ pg_reload_conf
+---------------------------------------------------------------------
+ t
+(1 row)
+
+\c - - - :worker_2_port
+ALTER SYSTEM SET citus.enable_ddl_propagation = 'true';
+SELECT pg_reload_conf();
+ pg_reload_conf
+---------------------------------------------------------------------
+ t
+(1 row)
+
+\c - - - :master_port
 SET citus.next_shard_id TO 1220000;
 ALTER SEQUENCE pg_catalog.pg_dist_colocationid_seq RESTART 1390000;
 ALTER SEQUENCE pg_catalog.pg_dist_groupid_seq RESTART 1;

src/test/regress/pg_regress_multi.pl

@@ -492,6 +492,7 @@ push(@pgOptions, "citus.stat_tenants_limit = 2");
 push(@pgOptions, "citus.stat_tenants_track = 'ALL'");
 push(@pgOptions, "citus.main_db = 'regression'");
 push(@pgOptions, "citus.superuser = 'postgres'");
+push(@pgOptions, "citus.enable_ddl_propagation=false");
 
 # Some tests look at shards in pg_class, make sure we can usually see them:
 push(@pgOptions, "citus.show_shards_for_app_name_prefixes='pg_regress'");

src/test/regress/sql/alter_database_from_nonmain_db.sql

@@ -1,5 +1,7 @@
 SET citus.superuser TO 'postgres';
 set citus.enable_create_database_propagation=on;
+set citus.log_remote_commands = true;
+set citus.grep_remote_commands = "%CREATE DATABASE%";
 create database test_alter_db_from_nonmain_db;
 create database "altered_database!'2";
 reset citus.enable_create_database_propagation;
@@ -18,7 +20,12 @@ CREATE TABLESPACE "ts-needs\!escape2" LOCATION :'alter_db_tablespace';
 CREATE TABLESPACE "ts-needs\!escape2" LOCATION :'alter_db_tablespace';
 
 \c test_alter_db_from_nonmain_db
+set citus.enable_ddl_propagation=true;
+
 set citus.log_remote_commands = true;
+set citus.grep_remote_commands = "%CREATE DATABASE%";
+create database test1;
+
 set citus.grep_remote_commands = "%ALTER DATABASE%";
 
 alter database "altered_database!'2" set tablespace "ts-needs\!escape2";
@@ -35,8 +42,10 @@ alter database "altered_database!'2" with
     IS_TEMPLATE false;
 
 \c regression
+set citus.enable_ddl_propagation=true;
 create role test_owner_non_main_db;
 \c test_alter_db_from_nonmain_db
+set citus.enable_ddl_propagation=true;
 set citus.log_remote_commands = true;
 set citus.grep_remote_commands = "%ALTER DATABASE%";
 set citus.enable_create_database_propagation=on;
@@ -70,6 +79,8 @@ alter database "altered_database!'2" set lock_timeout to DEFAULT;
 alter database "altered_database!'2" RESET lock_timeout;
 ALTER DATABASE "altered_database!'2" RESET ALL;
 \c regression
+show citus.enable_ddl_propagation;
+set citus.enable_ddl_propagation=true;
 set citus.enable_create_database_propagation=on;
 drop database "altered_database!'2";
 drop database test_alter_db_from_nonmain_db;

src/test/regress/sql/failure_setup.sql

@@ -1,3 +1,15 @@
+ALTER SYSTEM SET citus.enable_ddl_propagation = 'true';
+SELECT pg_reload_conf();
+
+\c - - - :worker_1_port
+ALTER SYSTEM SET citus.enable_ddl_propagation = 'true';
+SELECT pg_reload_conf();
+
+\c - - - :worker_2_port
+ALTER SYSTEM SET citus.enable_ddl_propagation = 'true';
+SELECT pg_reload_conf();
+
+\c - - - :master_port
 SELECT citus.mitmproxy('conn.allow()');
 
 -- add the workers

src/test/regress/sql/minimal_cluster_management.sql

@@ -1,3 +1,24 @@
+ALTER SYSTEM SET citus.enable_ddl_propagation = 'true';
+SELECT pg_reload_conf();
+
+\c - - - :worker_1_port
+ALTER SYSTEM SET citus.enable_ddl_propagation = 'true';
+SELECT pg_reload_conf();
+
+\c - - - :worker_2_port
+ALTER SYSTEM SET citus.enable_ddl_propagation = 'true';
+SELECT pg_reload_conf();
+
+\c - - - :master_port
+\c - - - :worker_1_port
+ALTER SYSTEM SET citus.enable_ddl_propagation = 'true';
+SELECT pg_reload_conf();
+
+\c - - - :worker_2_port
+ALTER SYSTEM SET citus.enable_ddl_propagation = 'true';
+SELECT pg_reload_conf();
+
+\c - - - :master_port
 SET citus.next_shard_id TO 1220000;
 ALTER SEQUENCE pg_catalog.pg_dist_colocationid_seq RESTART 1390000;
 ALTER SEQUENCE pg_catalog.pg_dist_groupid_seq RESTART 1;

src/test/regress/sql/multi_cluster_management.sql

@@ -1,3 +1,16 @@
+ALTER SYSTEM SET citus.enable_ddl_propagation = 'true';
+SELECT pg_reload_conf();
+
+\c - - - :worker_1_port
+ALTER SYSTEM SET citus.enable_ddl_propagation = 'true';
+SELECT pg_reload_conf();
+
+\c - - - :worker_2_port
+ALTER SYSTEM SET citus.enable_ddl_propagation = 'true';
+SELECT pg_reload_conf();
+
+\c - - - :master_port
+
 SET citus.next_shard_id TO 1220000;
 ALTER SEQUENCE pg_catalog.pg_dist_colocationid_seq RESTART 1390000;
 ALTER SEQUENCE pg_catalog.pg_dist_groupid_seq RESTART 1;