Add failure handling for CREATE DATABASE commands (#7483)

In preprocess phase, we save the original database name, replace
dbname field of CreatedbStmt with a temporary name (to let Postgres
to create the database with the temporary name locally) and then
we insert a cleanup record for the temporary database name on all
nodes **(\*\*)**.

And in postprocess phase, we first rename the temporary database
back to its original name for local node and then return a list of
distributed DDL jobs i) to create the database with the temporary
name and then ii) to rename it back to its original name on other
nodes. That way, if CREATE DATABASE fails on any of the nodes, the
temporary database will be cleaned up by the cleanup records that
we inserted in preprocess phase and in case of a failure, we won't
leak any databases called as the name that user intended to use for
the database.

Solves the problem documented in
https://github.com/citusdata/citus/issues/7369
for CREATE DATABASE commands.

**(\*\*):** To ensure that we insert cleanup records on all nodes,
with this PR we also start requiring having the coordinator in the
metadata because otherwise we would skip inserting a cleanup record
for the coordinator.

src/backend/distributed/commands/database.c

@@ -40,15 +40,38 @@
 #include "distributed/deparse_shard_query.h"
 #include "distributed/deparser.h"
 #include "distributed/listutils.h"
+#include "distributed/local_executor.h"
 #include "distributed/metadata/distobject.h"
 #include "distributed/metadata_sync.h"
 #include "distributed/metadata_utility.h"
 #include "distributed/multi_executor.h"
 #include "distributed/relation_access_tracking.h"
 #include "distributed/serialize_distributed_ddls.h"
+#include "distributed/shard_cleaner.h"
 #include "distributed/worker_protocol.h"
 #include "distributed/worker_transaction.h"
 
+
+/*
+ * Used to save original name of the database before it is replaced with a
+ * temporary name for failure handling purposes in PreprocessCreateDatabaseStmt().
+ */
+static char *CreateDatabaseCommandOriginalDbName = NULL;
+
+
+/*
+ * The format string used when creating a temporary databases for failure
+ * handling purposes.
+ *
+ * The fields are as follows to ensure using a unique name for each temporary
+ * database:
+ * - operationId: The operation id returned by RegisterOperationNeedingCleanup().
+ * - groupId:     The group id of the worker node where CREATE DATABASE command
+ *                is issued from.
+ */
+#define TEMP_DATABASE_NAME_FMT "citus_temp_database_%lu_%d"
+
+
 /*
  * DatabaseCollationInfo is used to store collation related information of a database.
  */
@@ -286,8 +309,9 @@ PreprocessAlterDatabaseStmt(Node *node, const char *queryString,
 		 * NontransactionalNodeDDLTask to run the command on the workers outside
 		 * the transaction block.
 		 */
-
+		bool warnForPartialFailure = true;
-		return NontransactionalNodeDDLTaskList(NON_COORDINATOR_NODES, commands);
+		return NontransactionalNodeDDLTaskList(NON_COORDINATOR_NODES, commands,
+											   warnForPartialFailure);
 	}
 	else
 	{
@@ -453,7 +477,12 @@ PreprocessAlterDatabaseSetStmt(Node *node, const char *queryString,
  *
  * In this stage, we perform validations that we want to ensure before delegating to
  * previous utility hooks because it might not be convenient to throw an error in an
- * implicit transaction that creates a database.
+ * implicit transaction that creates a database. Also in this stage, we save the original
+ * database name and replace dbname field with a temporary name for failure handling
+ * purposes. We let Postgres create the database with the temporary name, insert a cleanup
+ * record for the temporary database name on all nodes and let PostprocessCreateDatabaseStmt()
+ * to return the distributed DDL job that both creates the database with the temporary name
+ * and then renames it back to its original name.
  *
  * We also serialize database commands globally by acquiring a Citus specific advisory
  * lock based on OCLASS_DATABASE on the first primary worker node.
@@ -467,22 +496,56 @@ PreprocessCreateDatabaseStmt(Node *node, const char *queryString,
 		return NIL;
 	}
 
-	EnsurePropagationToCoordinator();
+	EnsureCoordinatorIsInMetadata();
 
 	CreatedbStmt *stmt = castNode(CreatedbStmt, node);
 	EnsureSupportedCreateDatabaseCommand(stmt);
 
 	SerializeDistributedDDLsOnObjectClass(OCLASS_DATABASE);
 
+	OperationId operationId = RegisterOperationNeedingCleanup();
+
+	char *tempDatabaseName = psprintf(TEMP_DATABASE_NAME_FMT,
+									  operationId, GetLocalGroupId());
+
+	List *remoteNodes = TargetWorkerSetNodeList(ALL_SHARD_NODES, RowShareLock);
+	WorkerNode *remoteNode = NULL;
+	foreach_ptr(remoteNode, remoteNodes)
+	{
+		InsertCleanupRecordOutsideTransaction(
+			CLEANUP_OBJECT_DATABASE,
+			pstrdup(quote_identifier(tempDatabaseName)),
+			remoteNode->groupId,
+			CLEANUP_ON_FAILURE
+			);
+	}
+
+	CreateDatabaseCommandOriginalDbName = stmt->dbname;
+	stmt->dbname = tempDatabaseName;
+
+	/*
+	 * Delete cleanup records in the same transaction so that if the current
+	 * transactions fails for some reason, then the cleanup records won't be
+	 * deleted. In the happy path, we will delete the cleanup records without
+	 * deferring them to the background worker.
+	 */
+	FinalizeOperationNeedingCleanupOnSuccess("create database");
+
 	return NIL;
 }
 
 
 /*
  * PostprocessCreateDatabaseStmt is executed after the statement is applied to the local
- * postgres instance. In this stage we prepare the commands that need to be run on
+ * postgres instance.
- * all workers to create the database.
  *
+ * In this stage, we first rename the temporary database back to its original name for
+ * local node and then return a list of distributed DDL jobs to create the database with
+ * the temporary name and then to rename it back to its original name. That way, if CREATE
+ * DATABASE fails on any of the nodes, the temporary database will be cleaned up by the
+ * cleanup records that we inserted in PreprocessCreateDatabaseStmt() and in case of a
+ * failure, we won't leak any databases called as the name that user intended to use for
+ * the database.
  */
 List *
 PostprocessCreateDatabaseStmt(Node *node, const char *queryString)
@@ -515,9 +578,55 @@ PostprocessCreateDatabaseStmt(Node *node, const char *queryString)
 	 * block, we need to use NontransactionalNodeDDLTaskList() to send the CREATE
 	 * DATABASE statement to the workers.
 	 */
+	bool warnForPartialFailure = false;
 	List *createDatabaseDDLJobList =
-		NontransactionalNodeDDLTaskList(REMOTE_NODES, createDatabaseCommands);
+		NontransactionalNodeDDLTaskList(REMOTE_NODES, createDatabaseCommands,
-	return createDatabaseDDLJobList;
+										warnForPartialFailure);
+
+	CreatedbStmt *stmt = castNode(CreatedbStmt, node);
+
+	char *renameDatabaseCommand =
+		psprintf("ALTER DATABASE %s RENAME TO %s",
+				 quote_identifier(stmt->dbname),
+				 quote_identifier(CreateDatabaseCommandOriginalDbName));
+
+	List *renameDatabaseCommands = list_make3(DISABLE_DDL_PROPAGATION,
+											  renameDatabaseCommand,
+											  ENABLE_DDL_PROPAGATION);
+
+	/*
+	 * We use NodeDDLTaskList() to send the RENAME DATABASE statement to the
+	 * workers because we want to execute it in a coordinated transaction.
+	 */
+	List *renameDatabaseDDLJobList =
+		NodeDDLTaskList(REMOTE_NODES, renameDatabaseCommands);
+
+	/*
+	 * Temporarily disable citus.enable_ddl_propagation before issuing
+	 * rename command locally because we don't want to execute it on remote
+	 * nodes yet. We will execute it on remote nodes by returning it as a
+	 * distributed DDL job.
+	 *
+	 * The reason why we don't want to execute it on remote nodes yet is that
+	 * the database is not created on remote nodes yet.
+	 */
+	int saveNestLevel = NewGUCNestLevel();
+	set_config_option("citus.enable_ddl_propagation", "off",
+					  (superuser() ? PGC_SUSET : PGC_USERSET), PGC_S_SESSION,
+					  GUC_ACTION_LOCAL, true, 0, false);
+
+	ExecuteUtilityCommand(renameDatabaseCommand);
+
+	AtEOXact_GUC(true, saveNestLevel);
+
+	/*
+	 * Restore the original database name because MarkObjectDistributed()
+	 * resolves oid of the object based on the database name and is called
+	 * after executing the distributed DDL job that renames temporary database.
+	 */
+	stmt->dbname = CreateDatabaseCommandOriginalDbName;
+
+	return list_concat(createDatabaseDDLJobList, renameDatabaseDDLJobList);
 }
 
 
@@ -571,8 +680,10 @@ PreprocessDropDatabaseStmt(Node *node, const char *queryString,
 	 * use NontransactionalNodeDDLTaskList() to send the DROP DATABASE statement
 	 * to the workers.
 	 */
+	bool warnForPartialFailure = true;
 	List *dropDatabaseDDLJobList =
-		NontransactionalNodeDDLTaskList(REMOTE_NODES, dropDatabaseCommands);
+		NontransactionalNodeDDLTaskList(REMOTE_NODES, dropDatabaseCommands,
+										warnForPartialFailure);
 	return dropDatabaseDDLJobList;
 }
 

src/backend/distributed/commands/index.c

@@ -493,6 +493,7 @@ GenerateCreateIndexDDLJob(IndexStmt *createIndexStatement, const char *createInd
 	ddlJob->startNewTransaction = createIndexStatement->concurrent;
 	ddlJob->metadataSyncCommand = createIndexCommand;
 	ddlJob->taskList = CreateIndexTaskList(createIndexStatement);
+	ddlJob->warnForPartialFailure = true;
 
 	return ddlJob;
 }
@@ -652,6 +653,7 @@ PreprocessReindexStmt(Node *node, const char *reindexCommand,
 																	"concurrently");
 			ddlJob->metadataSyncCommand = reindexCommand;
 			ddlJob->taskList = CreateReindexTaskList(relationId, reindexStatement);
+			ddlJob->warnForPartialFailure = true;
 
 			ddlJobs = list_make1(ddlJob);
 		}
@@ -780,6 +782,7 @@ PreprocessDropIndexStmt(Node *node, const char *dropIndexCommand,
 		ddlJob->metadataSyncCommand = dropIndexCommand;
 		ddlJob->taskList = DropIndexTaskList(distributedRelationId, distributedIndexId,
 											 dropIndexStatement);
+		ddlJob->warnForPartialFailure = true;
 
 		ddlJobs = list_make1(ddlJob);
 	}

src/backend/distributed/commands/utility_hook.c

@@ -1377,7 +1377,7 @@ ExecuteDistributedDDLJob(DDLJob *ddlJob)
 						 errhint("Use DROP INDEX CONCURRENTLY IF EXISTS to remove the "
 								 "invalid index, then retry the original command.")));
 			}
-			else
+			else if (ddlJob->warnForPartialFailure)
 			{
 				ereport(WARNING,
 						(errmsg(
@@ -1386,9 +1386,9 @@ ExecuteDistributedDDLJob(DDLJob *ddlJob)
 							 "state.\nIf the problematic command is a CREATE operation, "
 							 "consider using the 'IF EXISTS' syntax to drop the object,"
 							 "\nif applicable, and then re-attempt the original command.")));
-
-				PG_RE_THROW();
 			}
+
+			PG_RE_THROW();
 		}
 		PG_END_TRY();
 	}
@@ -1604,9 +1604,12 @@ DDLTaskList(Oid relationId, const char *commandString)
  * NontransactionalNodeDDLTaskList builds a list of tasks to execute a DDL command on a
  * given target set of nodes with cannotBeExecutedInTransaction is set to make sure
  * that task list is executed outside a transaction block.
+ *
+ * Also sets warnForPartialFailure for the returned DDLJobs.
  */
 List *
-NontransactionalNodeDDLTaskList(TargetWorkerSet targets, List *commands)
+NontransactionalNodeDDLTaskList(TargetWorkerSet targets, List *commands,
+								bool warnForPartialFailure)
 {
 	List *ddlJobs = NodeDDLTaskList(targets, commands);
 	DDLJob *ddlJob = NULL;
@@ -1617,6 +1620,8 @@ NontransactionalNodeDDLTaskList(TargetWorkerSet targets, List *commands)
 		{
 			task->cannotBeExecutedInTransaction = true;
 		}
+
+		ddlJob->warnForPartialFailure = warnForPartialFailure;
 	}
 	return ddlJobs;
 }

src/backend/distributed/operations/shard_cleaner.c

@@ -92,6 +92,8 @@ static bool TryDropReplicationSlotOutsideTransaction(char *replicationSlotName,
 													 char *nodeName,
 													 int nodePort);
 static bool TryDropUserOutsideTransaction(char *username, char *nodeName, int nodePort);
+static bool TryDropDatabaseOutsideTransaction(char *databaseName, char *nodeName,
+											  int nodePort);
 
 static CleanupRecord * GetCleanupRecordByNameAndType(char *objectName,
 													 CleanupObject type);
@@ -141,7 +143,6 @@ Datum
 citus_cleanup_orphaned_resources(PG_FUNCTION_ARGS)
 {
 	CheckCitusVersion(ERROR);
-	EnsureCoordinator();
 	PreventInTransactionBlock(true, "citus_cleanup_orphaned_resources");
 
 	int droppedCount = DropOrphanedResourcesForCleanup();
@@ -245,12 +246,6 @@ TryDropOrphanedResources()
 static int
 DropOrphanedResourcesForCleanup()
 {
-	/* Only runs on Coordinator */
-	if (!IsCoordinator())
-	{
-		return 0;
-	}
-
 	List *cleanupRecordList = ListCleanupRecords();
 
 	/*
@@ -608,6 +603,12 @@ TryDropResourceByCleanupRecordOutsideTransaction(CleanupRecord *record,
 			return TryDropUserOutsideTransaction(record->objectName, nodeName, nodePort);
 		}
 
+		case CLEANUP_OBJECT_DATABASE:
+		{
+			return TryDropDatabaseOutsideTransaction(record->objectName, nodeName,
+													 nodePort);
+		}
+
 		default:
 		{
 			ereport(WARNING, (errmsg(
@@ -888,6 +889,69 @@ TryDropUserOutsideTransaction(char *username,
 }
 
 
+/*
+ * TryDropDatabaseOutsideTransaction drops the database with the given name
+ * if it exists.
+ */
+static bool
+TryDropDatabaseOutsideTransaction(char *databaseName, char *nodeName, int nodePort)
+{
+	int connectionFlags = (OUTSIDE_TRANSACTION | FORCE_NEW_CONNECTION);
+	MultiConnection *connection = GetNodeUserDatabaseConnection(connectionFlags,
+																nodeName, nodePort,
+																CitusExtensionOwnerName(),
+																NULL);
+
+	if (PQstatus(connection->pgConn) != CONNECTION_OK)
+	{
+		return false;
+	}
+
+	/*
+	 * We want to disable DDL propagation and set lock_timeout before issuing
+	 * the DROP DATABASE command but we cannot do so in a way that's scoped
+	 * to the DROP DATABASE command. This is because, we cannot use a
+	 * transaction block for the DROP DATABASE command.
+	 *
+	 * For this reason, to avoid leaking the lock_timeout and DDL propagation
+	 * settings to future commands, we force the connection to close at the end
+	 * of the transaction.
+	 */
+	ForceConnectionCloseAtTransactionEnd(connection);
+
+	/*
+	 * The DROP DATABASE command should not propagate, so we disable DDL
+	 * propagation.
+	 */
+	List *commandList = list_make3(
+		"SET lock_timeout TO '1s'",
+		"SET citus.enable_ddl_propagation TO OFF;",
+		psprintf("DROP DATABASE IF EXISTS %s;", quote_identifier(databaseName))
+		);
+
+	bool executeCommand = true;
+
+	const char *commandString = NULL;
+	foreach_ptr(commandString, commandList)
+	{
+		/*
+		 * Cannot use SendOptionalCommandListToWorkerOutsideTransactionWithConnection()
+		 * because we don't want to open a transaction block on remote nodes as DROP
+		 * DATABASE commands cannot be run inside a transaction block.
+		 */
+		if (ExecuteOptionalRemoteCommand(connection, commandString, NULL) !=
+			RESPONSE_OKAY)
+		{
+			executeCommand = false;
+			break;
+		}
+	}
+
+	CloseConnection(connection);
+	return executeCommand;
+}
+
+
 /*
  * ErrorIfCleanupRecordForShardExists errors out if a cleanup record for the given
  * shard name exists.

src/include/distributed/commands/utility_hook.h

@@ -75,6 +75,15 @@ typedef struct DDLJob
 	const char *metadataSyncCommand;
 
 	List *taskList;            /* worker DDL tasks to execute */
+
+	/*
+	 * Only applicable when any of the tasks cannot be executed in a
+	 * transaction block.
+	 *
+	 * Controls whether to emit a warning within the utility hook in case of a
+	 * failure.
+	 */
+	bool warnForPartialFailure;
 } DDLJob;
 
 extern ProcessUtility_hook_type PrevProcessUtility;
@@ -94,7 +103,8 @@ extern void ProcessUtilityParseTree(Node *node, const char *queryString,
 extern void MarkInvalidateForeignKeyGraph(void);
 extern void InvalidateForeignKeyGraphForDDL(void);
 extern List * DDLTaskList(Oid relationId, const char *commandString);
-extern List * NontransactionalNodeDDLTaskList(TargetWorkerSet targets, List *commands);
+extern List * NontransactionalNodeDDLTaskList(TargetWorkerSet targets, List *commands,
+											  bool warnForPartialFailure);
 extern List * NodeDDLTaskList(TargetWorkerSet targets, List *commands);
 extern bool AlterTableInProgress(void);
 extern bool DropSchemaOrDBInProgress(void);

src/include/distributed/shard_cleaner.h

@@ -41,7 +41,8 @@ typedef enum CleanupObject
 	CLEANUP_OBJECT_SUBSCRIPTION = 2,
 	CLEANUP_OBJECT_REPLICATION_SLOT = 3,
 	CLEANUP_OBJECT_PUBLICATION = 4,
-	CLEANUP_OBJECT_USER = 5
+	CLEANUP_OBJECT_USER = 5,
+	CLEANUP_OBJECT_DATABASE = 6
 } CleanupObject;
 
 /*

src/test/regress/expected/alter_database_propagation.out

@@ -140,7 +140,12 @@ DETAIL:  on server postgres@localhost:xxxxx connectionId: xxxxxxx
 NOTICE:  issuing ALTER DATABASE regression RESET lock_timeout
 DETAIL:  on server postgres@localhost:xxxxx connectionId: xxxxxxx
 set citus.enable_create_database_propagation=on;
+SET citus.next_operation_id TO 3000;
 create database "regression!'2";
+NOTICE:  issuing ALTER DATABASE citus_temp_database_3000_0 RENAME TO "regression!'2"
+DETAIL:  on server postgres@localhost:xxxxx connectionId: xxxxxxx
+NOTICE:  issuing ALTER DATABASE citus_temp_database_3000_0 RENAME TO "regression!'2"
+DETAIL:  on server postgres@localhost:xxxxx connectionId: xxxxxxx
 alter database "regression!'2" with CONNECTION LIMIT 100;
 NOTICE:  issuing ALTER DATABASE "regression!'2" WITH CONNECTION LIMIT 100;
 DETAIL:  on server postgres@localhost:xxxxx connectionId: xxxxxxx
@@ -189,7 +194,12 @@ alter DATABASE local_regression rename to local_regression2;
 drop database local_regression2;
 set citus.enable_create_database_propagation=on;
 drop database regression3;
+SET citus.next_operation_id TO 3100;
 create database "regression!'4";
+NOTICE:  issuing ALTER DATABASE citus_temp_database_3100_0 RENAME TO "regression!'4"
+DETAIL:  on server postgres@localhost:xxxxx connectionId: xxxxxxx
+NOTICE:  issuing ALTER DATABASE citus_temp_database_3100_0 RENAME TO "regression!'4"
+DETAIL:  on server postgres@localhost:xxxxx connectionId: xxxxxxx
 SELECT result FROM run_command_on_all_nodes(
   $$
   ALTER TABLESPACE alter_db_tablespace RENAME TO "ts-needs\!escape"

src/test/regress/expected/create_drop_database_propagation.out

@@ -427,11 +427,16 @@ SELECT * FROM public.check_database_on_all_nodes('my_template_database') ORDER B
 --tests for special characters in database name
 set citus.enable_create_database_propagation=on;
 SET citus.log_remote_commands = true;
-set citus.grep_remote_commands = '%CREATE DATABASE%';
+set citus.grep_remote_commands = '%DATABASE%';
+SET citus.next_operation_id TO 2000;
 create database "mydatabase#1'2";
-NOTICE:  issuing CREATE DATABASE "mydatabase#1'2"
+NOTICE:  issuing CREATE DATABASE citus_temp_database_2000_0
 DETAIL:  on server postgres@localhost:xxxxx connectionId: xxxxxxx
-NOTICE:  issuing CREATE DATABASE "mydatabase#1'2"
+NOTICE:  issuing CREATE DATABASE citus_temp_database_2000_0
+DETAIL:  on server postgres@localhost:xxxxx connectionId: xxxxxxx
+NOTICE:  issuing ALTER DATABASE citus_temp_database_2000_0 RENAME TO "mydatabase#1'2"
+DETAIL:  on server postgres@localhost:xxxxx connectionId: xxxxxxx
+NOTICE:  issuing ALTER DATABASE citus_temp_database_2000_0 RENAME TO "mydatabase#1'2"
 DETAIL:  on server postgres@localhost:xxxxx connectionId: xxxxxxx
 set citus.grep_remote_commands = '%DROP DATABASE%';
 drop database if exists "mydatabase#1'2";
@@ -1264,6 +1269,95 @@ SELECT 1 FROM run_command_on_all_nodes($$REVOKE ALL ON TABLESPACE pg_default FRO
 
 DROP DATABASE no_createdb;
 DROP USER no_createdb;
+-- Test a failure scenario by trying to create a distributed database that
+-- already exists on one of the nodes.
+\c - - - :worker_1_port
+CREATE DATABASE "test_\!failure";
+NOTICE:  Citus partially supports CREATE DATABASE for distributed databases
+DETAIL:  Citus does not propagate CREATE DATABASE command to other nodes
+HINT:  You can manually create a database and its extensions on other nodes.
+\c - - - :master_port
+SET citus.enable_create_database_propagation TO ON;
+CREATE DATABASE "test_\!failure";
+ERROR:  database "test_\!failure" already exists
+CONTEXT:  while executing command on localhost:xxxxx
+SET client_min_messages TO WARNING;
+CALL citus_cleanup_orphaned_resources();
+RESET client_min_messages;
+SELECT result AS database_cleanedup_on_node FROM run_command_on_all_nodes($$SELECT COUNT(*)=0 FROM pg_database WHERE datname LIKE 'citus_temp_database_%'$$);
+ database_cleanedup_on_node
+---------------------------------------------------------------------
+ t
+ t
+ t
+(3 rows)
+
+SELECT * FROM public.check_database_on_all_nodes($$test_\!failure$$) ORDER BY node_type, result;
+      node_type       |                                                                                                                                                                                                                          result
+---------------------------------------------------------------------
+ coordinator (local)  | {"database_properties": null, "pg_dist_object_record_for_db_exists": false, "stale_pg_dist_object_record_for_a_db_exists": false}
+ worker node (remote) | {"database_properties": null, "pg_dist_object_record_for_db_exists": false, "stale_pg_dist_object_record_for_a_db_exists": false}
+ worker node (remote) | {"database_properties": {"datacl": null, "datname": "test_\\!failure", "datctype": "C", "encoding": "UTF8", "datcollate": "C", "tablespace": "pg_default", "daticurules": null, "datallowconn": true, "datconnlimit": -1, "daticulocale": null, "datistemplate": false, "database_owner": "postgres", "datcollversion": null, "datlocprovider": "c"}, "pg_dist_object_record_for_db_exists": false, "stale_pg_dist_object_record_for_a_db_exists": false}
+(3 rows)
+
+SET citus.enable_create_database_propagation TO OFF;
+CREATE DATABASE "test_\!failure1";
+NOTICE:  Citus partially supports CREATE DATABASE for distributed databases
+DETAIL:  Citus does not propagate CREATE DATABASE command to other nodes
+HINT:  You can manually create a database and its extensions on other nodes.
+\c - - - :worker_1_port
+DROP DATABASE "test_\!failure";
+SET citus.enable_create_database_propagation TO ON;
+CREATE DATABASE "test_\!failure1";
+ERROR:  database "test_\!failure1" already exists
+CONTEXT:  while executing command on localhost:xxxxx
+SET client_min_messages TO WARNING;
+CALL citus_cleanup_orphaned_resources();
+RESET client_min_messages;
+SELECT result AS database_cleanedup_on_node FROM run_command_on_all_nodes($$SELECT COUNT(*)=0 FROM pg_database WHERE datname LIKE 'citus_temp_database_%'$$);
+ database_cleanedup_on_node
+---------------------------------------------------------------------
+ t
+ t
+ t
+(3 rows)
+
+SELECT * FROM public.check_database_on_all_nodes($$test_\!failure1$$) ORDER BY node_type, result;
+      node_type       |                                                                                                                                                                                                                           result
+---------------------------------------------------------------------
+ coordinator (remote) | {"database_properties": {"datacl": null, "datname": "test_\\!failure1", "datctype": "C", "encoding": "UTF8", "datcollate": "C", "tablespace": "pg_default", "daticurules": null, "datallowconn": true, "datconnlimit": -1, "daticulocale": null, "datistemplate": false, "database_owner": "postgres", "datcollversion": null, "datlocprovider": "c"}, "pg_dist_object_record_for_db_exists": false, "stale_pg_dist_object_record_for_a_db_exists": false}
+ worker node (local)  | {"database_properties": null, "pg_dist_object_record_for_db_exists": false, "stale_pg_dist_object_record_for_a_db_exists": false}
+ worker node (remote) | {"database_properties": null, "pg_dist_object_record_for_db_exists": false, "stale_pg_dist_object_record_for_a_db_exists": false}
+(3 rows)
+
+\c - - - :master_port
+-- Before dropping local "test_\!failure1" database, test a failure scenario
+-- by trying to create a distributed database that already exists "on local
+-- node" this time.
+SET citus.enable_create_database_propagation TO ON;
+CREATE DATABASE "test_\!failure1";
+ERROR:  database "test_\!failure1" already exists
+SET client_min_messages TO WARNING;
+CALL citus_cleanup_orphaned_resources();
+RESET client_min_messages;
+SELECT result AS database_cleanedup_on_node FROM run_command_on_all_nodes($$SELECT COUNT(*)=0 FROM pg_database WHERE datname LIKE 'citus_temp_database_%'$$);
+ database_cleanedup_on_node
+---------------------------------------------------------------------
+ t
+ t
+ t
+(3 rows)
+
+SELECT * FROM public.check_database_on_all_nodes($$test_\!failure1$$) ORDER BY node_type, result;
+      node_type       |                                                                                                                                                                                                                           result
+---------------------------------------------------------------------
+ coordinator (local)  | {"database_properties": {"datacl": null, "datname": "test_\\!failure1", "datctype": "C", "encoding": "UTF8", "datcollate": "C", "tablespace": "pg_default", "daticurules": null, "datallowconn": true, "datconnlimit": -1, "daticulocale": null, "datistemplate": false, "database_owner": "postgres", "datcollversion": null, "datlocprovider": "c"}, "pg_dist_object_record_for_db_exists": false, "stale_pg_dist_object_record_for_a_db_exists": false}
+ worker node (remote) | {"database_properties": null, "pg_dist_object_record_for_db_exists": false, "stale_pg_dist_object_record_for_a_db_exists": false}
+ worker node (remote) | {"database_properties": null, "pg_dist_object_record_for_db_exists": false, "stale_pg_dist_object_record_for_a_db_exists": false}
+(3 rows)
+
+SET citus.enable_create_database_propagation TO OFF;
+DROP DATABASE "test_\!failure1";
 SET citus.enable_create_database_propagation TO ON;
 --clean up resources created by this test
 -- DROP TABLESPACE is not supported, so we need to drop it manually.

src/test/regress/expected/failure_create_database.out

@@ -0,0 +1,386 @@
+SET citus.enable_create_database_propagation TO ON;
+SET client_min_messages TO WARNING;
+SELECT 1 FROM citus_add_node('localhost', :master_port, 0);
+ ?column?
+---------------------------------------------------------------------
+        1
+(1 row)
+
+CREATE FUNCTION get_temp_databases_on_nodes()
+RETURNS TEXT AS $func$
+  SELECT array_agg(DISTINCT result ORDER BY result) AS temp_databases_on_nodes FROM run_command_on_all_nodes($$SELECT datname FROM pg_database WHERE datname LIKE 'citus_temp_database_%'$$) WHERE result != '';
+$func$
+LANGUAGE sql;
+CREATE FUNCTION count_db_cleanup_records()
+RETURNS TABLE(object_name TEXT, count INTEGER) AS $func$
+  SELECT object_name, COUNT(*) FROM pg_dist_cleanup WHERE object_name LIKE 'citus_temp_database_%' GROUP BY object_name;
+$func$
+LANGUAGE sql;
+CREATE FUNCTION ensure_no_temp_databases_on_any_nodes()
+RETURNS BOOLEAN AS $func$
+  SELECT bool_and(result::boolean) AS no_temp_databases_on_any_nodes FROM run_command_on_all_nodes($$SELECT COUNT(*)=0 FROM pg_database WHERE datname LIKE 'citus_temp_database_%'$$);
+$func$
+LANGUAGE sql;
+-- cleanup any orphaned resources from previous runs
+CALL citus_cleanup_orphaned_resources();
+SET citus.next_operation_id TO 4000;
+ALTER SYSTEM SET citus.defer_shard_delete_interval TO -1;
+SELECT pg_reload_conf();
+ pg_reload_conf
+---------------------------------------------------------------------
+ t
+(1 row)
+
+SELECT pg_sleep(0.1);
+ pg_sleep
+---------------------------------------------------------------------
+
+(1 row)
+
+SELECT citus.mitmproxy('conn.kill()');
+ mitmproxy
+---------------------------------------------------------------------
+
+(1 row)
+
+CREATE DATABASE db1;
+ERROR:  connection to the remote node postgres@localhost:xxxxx failed with the following error: connection not open
+SELECT citus.mitmproxy('conn.allow()');
+ mitmproxy
+---------------------------------------------------------------------
+
+(1 row)
+
+SELECT get_temp_databases_on_nodes();
+ get_temp_databases_on_nodes
+---------------------------------------------------------------------
+
+(1 row)
+
+SELECT * FROM count_db_cleanup_records();
+ object_name | count
+---------------------------------------------------------------------
+(0 rows)
+
+CALL citus_cleanup_orphaned_resources();
+SELECT ensure_no_temp_databases_on_any_nodes();
+ ensure_no_temp_databases_on_any_nodes
+---------------------------------------------------------------------
+ t
+(1 row)
+
+SELECT * FROM public.check_database_on_all_nodes($$db1$$) ORDER BY node_type, result;
+      node_type       |                                                              result
+---------------------------------------------------------------------
+ coordinator (local)  | {"database_properties": null, "pg_dist_object_record_for_db_exists": false, "stale_pg_dist_object_record_for_a_db_exists": false}
+ worker node (remote) | {"database_properties": null, "pg_dist_object_record_for_db_exists": false, "stale_pg_dist_object_record_for_a_db_exists": false}
+ worker node (remote) | {"database_properties": null, "pg_dist_object_record_for_db_exists": false, "stale_pg_dist_object_record_for_a_db_exists": false}
+(3 rows)
+
+SELECT citus.mitmproxy('conn.onQuery(query="^CREATE DATABASE").cancel(' || pg_backend_pid() || ')');
+ mitmproxy
+---------------------------------------------------------------------
+
+(1 row)
+
+CREATE DATABASE db1;
+ERROR:  canceling statement due to user request
+SELECT citus.mitmproxy('conn.allow()');
+ mitmproxy
+---------------------------------------------------------------------
+
+(1 row)
+
+SELECT get_temp_databases_on_nodes();
+ get_temp_databases_on_nodes
+---------------------------------------------------------------------
+ {citus_temp_database_4000_0}
+(1 row)
+
+SELECT * FROM count_db_cleanup_records();
+        object_name         | count
+---------------------------------------------------------------------
+ citus_temp_database_4000_0 |     3
+(1 row)
+
+CALL citus_cleanup_orphaned_resources();
+SELECT ensure_no_temp_databases_on_any_nodes();
+ ensure_no_temp_databases_on_any_nodes
+---------------------------------------------------------------------
+ t
+(1 row)
+
+SELECT * FROM public.check_database_on_all_nodes($$db1$$) ORDER BY node_type, result;
+      node_type       |                                                              result
+---------------------------------------------------------------------
+ coordinator (local)  | {"database_properties": null, "pg_dist_object_record_for_db_exists": false, "stale_pg_dist_object_record_for_a_db_exists": false}
+ worker node (remote) | {"database_properties": null, "pg_dist_object_record_for_db_exists": false, "stale_pg_dist_object_record_for_a_db_exists": false}
+ worker node (remote) | {"database_properties": null, "pg_dist_object_record_for_db_exists": false, "stale_pg_dist_object_record_for_a_db_exists": false}
+(3 rows)
+
+SELECT citus.mitmproxy('conn.onQuery(query="^ALTER DATABASE").cancel(' || pg_backend_pid() || ')');
+ mitmproxy
+---------------------------------------------------------------------
+
+(1 row)
+
+CREATE DATABASE db1;
+ERROR:  canceling statement due to user request
+SELECT citus.mitmproxy('conn.allow()');
+ mitmproxy
+---------------------------------------------------------------------
+
+(1 row)
+
+SELECT get_temp_databases_on_nodes();
+ get_temp_databases_on_nodes
+---------------------------------------------------------------------
+ {citus_temp_database_4001_0}
+(1 row)
+
+SELECT * FROM count_db_cleanup_records();
+        object_name         | count
+---------------------------------------------------------------------
+ citus_temp_database_4001_0 |     3
+(1 row)
+
+CALL citus_cleanup_orphaned_resources();
+SELECT ensure_no_temp_databases_on_any_nodes();
+ ensure_no_temp_databases_on_any_nodes
+---------------------------------------------------------------------
+ t
+(1 row)
+
+SELECT * FROM public.check_database_on_all_nodes($$db1$$) ORDER BY node_type, result;
+      node_type       |                                                              result
+---------------------------------------------------------------------
+ coordinator (local)  | {"database_properties": null, "pg_dist_object_record_for_db_exists": false, "stale_pg_dist_object_record_for_a_db_exists": false}
+ worker node (remote) | {"database_properties": null, "pg_dist_object_record_for_db_exists": false, "stale_pg_dist_object_record_for_a_db_exists": false}
+ worker node (remote) | {"database_properties": null, "pg_dist_object_record_for_db_exists": false, "stale_pg_dist_object_record_for_a_db_exists": false}
+(3 rows)
+
+SELECT citus.mitmproxy('conn.onQuery(query="^BEGIN TRANSACTION ISOLATION LEVEL READ COMMITTED").kill()');
+ mitmproxy
+---------------------------------------------------------------------
+
+(1 row)
+
+CREATE DATABASE db1;
+ERROR:  connection to the remote node postgres@localhost:xxxxx failed with the following error: connection not open
+SELECT citus.mitmproxy('conn.allow()');
+ mitmproxy
+---------------------------------------------------------------------
+
+(1 row)
+
+SELECT get_temp_databases_on_nodes();
+ get_temp_databases_on_nodes
+---------------------------------------------------------------------
+
+(1 row)
+
+SELECT * FROM count_db_cleanup_records();
+ object_name | count
+---------------------------------------------------------------------
+(0 rows)
+
+CALL citus_cleanup_orphaned_resources();
+SELECT ensure_no_temp_databases_on_any_nodes();
+ ensure_no_temp_databases_on_any_nodes
+---------------------------------------------------------------------
+ t
+(1 row)
+
+SELECT * FROM public.check_database_on_all_nodes($$db1$$) ORDER BY node_type, result;
+      node_type       |                                                              result
+---------------------------------------------------------------------
+ coordinator (local)  | {"database_properties": null, "pg_dist_object_record_for_db_exists": false, "stale_pg_dist_object_record_for_a_db_exists": false}
+ worker node (remote) | {"database_properties": null, "pg_dist_object_record_for_db_exists": false, "stale_pg_dist_object_record_for_a_db_exists": false}
+ worker node (remote) | {"database_properties": null, "pg_dist_object_record_for_db_exists": false, "stale_pg_dist_object_record_for_a_db_exists": false}
+(3 rows)
+
+SELECT citus.mitmproxy('conn.onQuery(query="^PREPARE TRANSACTION").kill()');
+ mitmproxy
+---------------------------------------------------------------------
+
+(1 row)
+
+CREATE DATABASE db1;
+ERROR:  connection not open
+CONTEXT:  while executing command on localhost:xxxxx
+SELECT citus.mitmproxy('conn.allow()');
+ mitmproxy
+---------------------------------------------------------------------
+
+(1 row)
+
+SELECT get_temp_databases_on_nodes();
+ get_temp_databases_on_nodes
+---------------------------------------------------------------------
+ {citus_temp_database_4002_0}
+(1 row)
+
+SELECT * FROM count_db_cleanup_records();
+        object_name         | count
+---------------------------------------------------------------------
+ citus_temp_database_4002_0 |     3
+(1 row)
+
+CALL citus_cleanup_orphaned_resources();
+SELECT ensure_no_temp_databases_on_any_nodes();
+ ensure_no_temp_databases_on_any_nodes
+---------------------------------------------------------------------
+ t
+(1 row)
+
+SELECT * FROM public.check_database_on_all_nodes($$db1$$) ORDER BY node_type, result;
+      node_type       |                                                              result
+---------------------------------------------------------------------
+ coordinator (local)  | {"database_properties": null, "pg_dist_object_record_for_db_exists": false, "stale_pg_dist_object_record_for_a_db_exists": false}
+ worker node (remote) | {"database_properties": null, "pg_dist_object_record_for_db_exists": false, "stale_pg_dist_object_record_for_a_db_exists": false}
+ worker node (remote) | {"database_properties": null, "pg_dist_object_record_for_db_exists": false, "stale_pg_dist_object_record_for_a_db_exists": false}
+(3 rows)
+
+SELECT citus.mitmproxy('conn.onQuery(query="^COMMIT PREPARED").kill()');
+ mitmproxy
+---------------------------------------------------------------------
+
+(1 row)
+
+CREATE DATABASE db1;
+WARNING:  connection not open
+CONTEXT:  while executing command on localhost:xxxxx
+WARNING:  failed to commit transaction on localhost:xxxxx
+SELECT citus.mitmproxy('conn.allow()');
+ mitmproxy
+---------------------------------------------------------------------
+
+(1 row)
+
+-- not call citus_cleanup_orphaned_resources() but recover the prepared transactions this time
+SELECT 1 FROM recover_prepared_transactions();
+ ?column?
+---------------------------------------------------------------------
+        1
+(1 row)
+
+SELECT ensure_no_temp_databases_on_any_nodes();
+ ensure_no_temp_databases_on_any_nodes
+---------------------------------------------------------------------
+ t
+(1 row)
+
+SELECT * FROM public.check_database_on_all_nodes($$db1$$) ORDER BY node_type, result;
+      node_type       |                                                                                                                                                                                                                    result
+---------------------------------------------------------------------
+ coordinator (local)  | {"database_properties": {"datacl": null, "datname": "db1", "datctype": "C", "encoding": "UTF8", "datcollate": "C", "tablespace": "pg_default", "daticurules": null, "datallowconn": true, "datconnlimit": -1, "daticulocale": null, "datistemplate": false, "database_owner": "postgres", "datcollversion": null, "datlocprovider": "c"}, "pg_dist_object_record_for_db_exists": true, "stale_pg_dist_object_record_for_a_db_exists": false}
+ worker node (remote) | {"database_properties": {"datacl": null, "datname": "db1", "datctype": "C", "encoding": "UTF8", "datcollate": "C", "tablespace": "pg_default", "daticurules": null, "datallowconn": true, "datconnlimit": -1, "daticulocale": null, "datistemplate": false, "database_owner": "postgres", "datcollversion": null, "datlocprovider": "c"}, "pg_dist_object_record_for_db_exists": true, "stale_pg_dist_object_record_for_a_db_exists": false}
+ worker node (remote) | {"database_properties": {"datacl": null, "datname": "db1", "datctype": "C", "encoding": "UTF8", "datcollate": "C", "tablespace": "pg_default", "daticurules": null, "datallowconn": true, "datconnlimit": -1, "daticulocale": null, "datistemplate": false, "database_owner": "postgres", "datcollversion": null, "datlocprovider": "c"}, "pg_dist_object_record_for_db_exists": true, "stale_pg_dist_object_record_for_a_db_exists": false}
+(3 rows)
+
+DROP DATABASE db1;
+-- after recovering the prepared transactions, cleanup records should also be removed
+SELECT * FROM count_db_cleanup_records();
+ object_name | count
+---------------------------------------------------------------------
+(0 rows)
+
+SELECT citus.mitmproxy('conn.onQuery(query="^SELECT citus_internal.acquire_citus_advisory_object_class_lock").kill()');
+ mitmproxy
+---------------------------------------------------------------------
+
+(1 row)
+
+CREATE DATABASE db1;
+ERROR:  connection to the remote node postgres@localhost:xxxxx failed with the following error: connection not open
+SELECT citus.mitmproxy('conn.allow()');
+ mitmproxy
+---------------------------------------------------------------------
+
+(1 row)
+
+SELECT get_temp_databases_on_nodes();
+ get_temp_databases_on_nodes
+---------------------------------------------------------------------
+
+(1 row)
+
+SELECT * FROM count_db_cleanup_records();
+ object_name | count
+---------------------------------------------------------------------
+(0 rows)
+
+CALL citus_cleanup_orphaned_resources();
+SELECT ensure_no_temp_databases_on_any_nodes();
+ ensure_no_temp_databases_on_any_nodes
+---------------------------------------------------------------------
+ t
+(1 row)
+
+SELECT * FROM public.check_database_on_all_nodes($$db1$$) ORDER BY node_type, result;
+      node_type       |                                                              result
+---------------------------------------------------------------------
+ coordinator (local)  | {"database_properties": null, "pg_dist_object_record_for_db_exists": false, "stale_pg_dist_object_record_for_a_db_exists": false}
+ worker node (remote) | {"database_properties": null, "pg_dist_object_record_for_db_exists": false, "stale_pg_dist_object_record_for_a_db_exists": false}
+ worker node (remote) | {"database_properties": null, "pg_dist_object_record_for_db_exists": false, "stale_pg_dist_object_record_for_a_db_exists": false}
+(3 rows)
+
+SELECT citus.mitmproxy('conn.onParse(query="^WITH distributed_object_data").kill()');
+ mitmproxy
+---------------------------------------------------------------------
+
+(1 row)
+
+CREATE DATABASE db1;
+ERROR:  connection not open
+CONTEXT:  while executing command on localhost:xxxxx
+SELECT citus.mitmproxy('conn.allow()');
+ mitmproxy
+---------------------------------------------------------------------
+
+(1 row)
+
+SELECT get_temp_databases_on_nodes();
+ get_temp_databases_on_nodes
+---------------------------------------------------------------------
+ {citus_temp_database_4004_0}
+(1 row)
+
+SELECT * FROM count_db_cleanup_records();
+        object_name         | count
+---------------------------------------------------------------------
+ citus_temp_database_4004_0 |     3
+(1 row)
+
+CALL citus_cleanup_orphaned_resources();
+SELECT ensure_no_temp_databases_on_any_nodes();
+ ensure_no_temp_databases_on_any_nodes
+---------------------------------------------------------------------
+ t
+(1 row)
+
+SELECT * FROM public.check_database_on_all_nodes($$db1$$) ORDER BY node_type, result;
+      node_type       |                                                              result
+---------------------------------------------------------------------
+ coordinator (local)  | {"database_properties": null, "pg_dist_object_record_for_db_exists": false, "stale_pg_dist_object_record_for_a_db_exists": false}
+ worker node (remote) | {"database_properties": null, "pg_dist_object_record_for_db_exists": false, "stale_pg_dist_object_record_for_a_db_exists": false}
+ worker node (remote) | {"database_properties": null, "pg_dist_object_record_for_db_exists": false, "stale_pg_dist_object_record_for_a_db_exists": false}
+(3 rows)
+
+CREATE DATABASE db1;
+-- show that a successful database creation doesn't leave any pg_dist_cleanup records behind
+SELECT * FROM count_db_cleanup_records();
+ object_name | count
+---------------------------------------------------------------------
+(0 rows)
+
+DROP DATABASE db1;
+DROP FUNCTION get_temp_databases_on_nodes();
+DROP FUNCTION ensure_no_temp_databases_on_any_nodes();
+DROP FUNCTION count_db_cleanup_records();
+SELECT 1 FROM citus_remove_node('localhost', :master_port);
+ ?column?
+---------------------------------------------------------------------
+        1
+(1 row)
+

src/test/regress/expected/isolation_database_cmd_from_any_node.out

@@ -1,6 +1,11 @@
 Parsed test spec with 2 sessions
 
 starting permutation: s1-begin s2-begin s1-acquire-citus-adv-oclass-lock s2-acquire-citus-adv-oclass-lock s1-commit s2-commit
+?column?
+---------------------------------------------------------------------
+       1
+(1 row)
+
 step s1-begin: BEGIN;
 step s2-begin: BEGIN;
 step s1-acquire-citus-adv-oclass-lock: SELECT citus_internal.acquire_citus_advisory_object_class_lock(value, NULL) FROM oclass_database;
@@ -18,8 +23,18 @@ acquire_citus_advisory_object_class_lock
 (1 row)
 
 step s2-commit: COMMIT;
+?column?
+---------------------------------------------------------------------
+       1
+(1 row)
+
 
 starting permutation: s1-create-testdb1 s1-begin s2-begin s1-acquire-citus-adv-oclass-lock-with-oid-testdb1 s2-acquire-citus-adv-oclass-lock-with-oid-testdb1 s1-commit s2-commit s1-drop-testdb1
+?column?
+---------------------------------------------------------------------
+       1
+(1 row)
+
 step s1-create-testdb1: CREATE DATABASE testdb1;
 step s1-begin: BEGIN;
 step s2-begin: BEGIN;
@@ -39,8 +54,18 @@ acquire_citus_advisory_object_class_lock
 
 step s2-commit: COMMIT;
 step s1-drop-testdb1: DROP DATABASE testdb1;
+?column?
+---------------------------------------------------------------------
+       1
+(1 row)
+
 
 starting permutation: s1-create-testdb1 s2-create-testdb2 s1-begin s2-begin s1-acquire-citus-adv-oclass-lock-with-oid-testdb1 s2-acquire-citus-adv-oclass-lock-with-oid-testdb2 s1-commit s2-commit s1-drop-testdb1 s2-drop-testdb2
+?column?
+---------------------------------------------------------------------
+       1
+(1 row)
+
 step s1-create-testdb1: CREATE DATABASE testdb1;
 step s2-create-testdb2: CREATE DATABASE testdb2;
 step s1-begin: BEGIN;
@@ -61,8 +86,18 @@ step s1-commit: COMMIT;
 step s2-commit: COMMIT;
 step s1-drop-testdb1: DROP DATABASE testdb1;
 step s2-drop-testdb2: DROP DATABASE testdb2;
+?column?
+---------------------------------------------------------------------
+       1
+(1 row)
+
 
 starting permutation: s2-create-testdb2 s1-begin s2-begin s1-acquire-citus-adv-oclass-lock s2-acquire-citus-adv-oclass-lock-with-oid-testdb2 s1-commit s2-commit s2-drop-testdb2
+?column?
+---------------------------------------------------------------------
+       1
+(1 row)
+
 step s2-create-testdb2: CREATE DATABASE testdb2;
 step s1-begin: BEGIN;
 step s2-begin: BEGIN;
@@ -81,8 +116,18 @@ acquire_citus_advisory_object_class_lock
 step s1-commit: COMMIT;
 step s2-commit: COMMIT;
 step s2-drop-testdb2: DROP DATABASE testdb2;
+?column?
+---------------------------------------------------------------------
+       1
+(1 row)
+
 
 starting permutation: s2-create-testdb2 s2-begin s2-alter-testdb2-set-lc_monetary s1-create-db1 s2-rollback s2-drop-testdb2 s1-drop-db1
+?column?
+---------------------------------------------------------------------
+       1
+(1 row)
+
 step s2-create-testdb2: CREATE DATABASE testdb2;
 step s2-begin: BEGIN;
 step s2-alter-testdb2-set-lc_monetary: ALTER DATABASE testdb2 SET lc_monetary TO 'C';
@@ -90,8 +135,18 @@ step s1-create-db1: CREATE DATABASE db1;
 step s2-rollback: ROLLBACK;
 step s2-drop-testdb2: DROP DATABASE testdb2;
 step s1-drop-db1: DROP DATABASE db1;
+?column?
+---------------------------------------------------------------------
+       1
+(1 row)
+
 
 starting permutation: s2-create-testdb2 s2-begin s2-alter-testdb2-set-lc_monetary s1-create-user-dbuser s1-grant-on-testdb2-to-dbuser s2-rollback s2-drop-testdb2 s1-drop-user-dbuser
+?column?
+---------------------------------------------------------------------
+       1
+(1 row)
+
 step s2-create-testdb2: CREATE DATABASE testdb2;
 step s2-begin: BEGIN;
 step s2-alter-testdb2-set-lc_monetary: ALTER DATABASE testdb2 SET lc_monetary TO 'C';
@@ -100,8 +155,18 @@ step s1-grant-on-testdb2-to-dbuser: GRANT ALL ON DATABASE testdb2 TO dbuser;
 step s2-rollback: ROLLBACK;
 step s2-drop-testdb2: DROP DATABASE testdb2;
 step s1-drop-user-dbuser: DROP USER dbuser;
+?column?
+---------------------------------------------------------------------
+       1
+(1 row)
+
 
 starting permutation: s2-create-testdb2 s2-begin s2-alter-testdb2-set-lc_monetary s1-create-testdb1 s1-create-user-dbuser s1-grant-on-testdb1-to-dbuser s2-rollback s2-drop-testdb2 s1-drop-testdb1 s1-drop-user-dbuser
+?column?
+---------------------------------------------------------------------
+       1
+(1 row)
+
 step s2-create-testdb2: CREATE DATABASE testdb2;
 step s2-begin: BEGIN;
 step s2-alter-testdb2-set-lc_monetary: ALTER DATABASE testdb2 SET lc_monetary TO 'C';
@@ -112,8 +177,18 @@ step s2-rollback: ROLLBACK;
 step s2-drop-testdb2: DROP DATABASE testdb2;
 step s1-drop-testdb1: DROP DATABASE testdb1;
 step s1-drop-user-dbuser: DROP USER dbuser;
+?column?
+---------------------------------------------------------------------
+       1
+(1 row)
+
 
 starting permutation: s1-create-testdb1 s2-create-testdb2 s1-begin s2-begin s1-alter-testdb1-rename-to-db1 s2-alter-testdb2-rename-to-db1 s1-commit s2-rollback s1-drop-db1 s2-drop-testdb2
+?column?
+---------------------------------------------------------------------
+       1
+(1 row)
+
 step s1-create-testdb1: CREATE DATABASE testdb1;
 step s2-create-testdb2: CREATE DATABASE testdb2;
 step s1-begin: BEGIN;
@@ -126,8 +201,18 @@ ERROR:  database "db1" already exists
 step s2-rollback: ROLLBACK;
 step s1-drop-db1: DROP DATABASE db1;
 step s2-drop-testdb2: DROP DATABASE testdb2;
+?column?
+---------------------------------------------------------------------
+       1
+(1 row)
+
 
 starting permutation: s1-create-testdb1 s2-create-testdb2 s1-begin s2-begin s1-alter-testdb1-rename-to-db1 s2-alter-testdb2-rename-to-db1 s1-rollback s2-commit s1-drop-testdb1 s2-drop-db1
+?column?
+---------------------------------------------------------------------
+       1
+(1 row)
+
 step s1-create-testdb1: CREATE DATABASE testdb1;
 step s2-create-testdb2: CREATE DATABASE testdb2;
 step s1-begin: BEGIN;
@@ -139,8 +224,18 @@ step s2-alter-testdb2-rename-to-db1: <... completed>
 step s2-commit: COMMIT;
 step s1-drop-testdb1: DROP DATABASE testdb1;
 step s2-drop-db1: DROP DATABASE db1;
+?column?
+---------------------------------------------------------------------
+       1
+(1 row)
+
 
 starting permutation: s1-create-testdb1 s1-begin s2-begin s1-alter-testdb1-rename-to-db1 s2-alter-testdb1-rename-to-db1 s1-commit s2-rollback s1-drop-db1
+?column?
+---------------------------------------------------------------------
+       1
+(1 row)
+
 step s1-create-testdb1: CREATE DATABASE testdb1;
 step s1-begin: BEGIN;
 step s2-begin: BEGIN;
@@ -151,8 +246,18 @@ step s2-alter-testdb1-rename-to-db1: <... completed>
 ERROR:  database "testdb1" does not exist
 step s2-rollback: ROLLBACK;
 step s1-drop-db1: DROP DATABASE db1;
+?column?
+---------------------------------------------------------------------
+       1
+(1 row)
+
 
 starting permutation: s1-create-testdb1 s1-begin s2-begin s1-alter-testdb1-rename-to-db1 s2-alter-testdb1-rename-to-db1 s1-rollback s2-commit s2-drop-db1
+?column?
+---------------------------------------------------------------------
+       1
+(1 row)
+
 step s1-create-testdb1: CREATE DATABASE testdb1;
 step s1-begin: BEGIN;
 step s2-begin: BEGIN;
@@ -162,8 +267,18 @@ step s1-rollback: ROLLBACK;
 step s2-alter-testdb1-rename-to-db1: <... completed>
 step s2-commit: COMMIT;
 step s2-drop-db1: DROP DATABASE db1;
+?column?
+---------------------------------------------------------------------
+       1
+(1 row)
+
 
 starting permutation: s2-create-testdb2 s2-begin s2-alter-testdb2-rename-to-db1 s1-create-db1 s2-rollback s2-drop-testdb2 s1-drop-db1
+?column?
+---------------------------------------------------------------------
+       1
+(1 row)
+
 step s2-create-testdb2: CREATE DATABASE testdb2;
 step s2-begin: BEGIN;
 step s2-alter-testdb2-rename-to-db1: ALTER DATABASE testdb2 RENAME TO db1;
@@ -172,8 +287,18 @@ step s2-rollback: ROLLBACK;
 step s1-create-db1: <... completed>
 step s2-drop-testdb2: DROP DATABASE testdb2;
 step s1-drop-db1: DROP DATABASE db1;
+?column?
+---------------------------------------------------------------------
+       1
+(1 row)
+
 
 starting permutation: s2-create-testdb2 s2-begin s2-alter-testdb2-rename-to-db1 s1-create-db1 s2-commit s2-drop-db1
+?column?
+---------------------------------------------------------------------
+       1
+(1 row)
+
 step s2-create-testdb2: CREATE DATABASE testdb2;
 step s2-begin: BEGIN;
 step s2-alter-testdb2-rename-to-db1: ALTER DATABASE testdb2 RENAME TO db1;
@@ -182,8 +307,18 @@ step s2-commit: COMMIT;
 step s1-create-db1: <... completed>
 ERROR:  database "db1" already exists
 step s2-drop-db1: DROP DATABASE db1;
+?column?
+---------------------------------------------------------------------
+       1
+(1 row)
+
 
 starting permutation: s2-create-testdb2 s2-begin s2-alter-testdb2-rename-to-db2 s1-create-db1 s2-commit s2-drop-db2 s1-drop-db1
+?column?
+---------------------------------------------------------------------
+       1
+(1 row)
+
 step s2-create-testdb2: CREATE DATABASE testdb2;
 step s2-begin: BEGIN;
 step s2-alter-testdb2-rename-to-db2: ALTER DATABASE testdb2 RENAME TO db2;
@@ -192,16 +327,36 @@ step s2-commit: COMMIT;
 step s1-create-db1: <... completed>
 step s2-drop-db2: DROP DATABASE db2;
 step s1-drop-db1: DROP DATABASE db1;
+?column?
+---------------------------------------------------------------------
+       1
+(1 row)
+
 
 starting permutation: s2-create-testdb2 s2-begin s2-alter-testdb2-rename-to-db1 s1-drop-testdb2 s2-rollback
+?column?
+---------------------------------------------------------------------
+       1
+(1 row)
+
 step s2-create-testdb2: CREATE DATABASE testdb2;
 step s2-begin: BEGIN;
 step s2-alter-testdb2-rename-to-db1: ALTER DATABASE testdb2 RENAME TO db1;
 step s1-drop-testdb2: DROP DATABASE testdb2; <waiting ...>
 step s2-rollback: ROLLBACK;
 step s1-drop-testdb2: <... completed>
+?column?
+---------------------------------------------------------------------
+       1
+(1 row)
+
 
 starting permutation: s2-create-testdb2 s1-create-db1 s2-begin s2-alter-testdb2-rename-to-db2 s1-drop-db1 s2-commit s2-drop-db2
+?column?
+---------------------------------------------------------------------
+       1
+(1 row)
+
 step s2-create-testdb2: CREATE DATABASE testdb2;
 step s1-create-db1: CREATE DATABASE db1;
 step s2-begin: BEGIN;
@@ -209,3 +364,8 @@ step s2-alter-testdb2-rename-to-db2: ALTER DATABASE testdb2 RENAME TO db2;
 step s1-drop-db1: DROP DATABASE db1;
 step s2-commit: COMMIT;
 step s2-drop-db2: DROP DATABASE db2;
+?column?
+---------------------------------------------------------------------
+       1
+(1 row)
+

src/test/regress/failure_schedule

@@ -35,6 +35,7 @@ test: failure_mx_metadata_sync
 test: failure_mx_metadata_sync_multi_trans
 test: failure_connection_establishment
 test: failure_non_main_db_2pc
+test: failure_create_database
 
 # this test syncs metadata to the workers
 test: failure_failover_to_local_execution

src/test/regress/spec/isolation_database_cmd_from_any_node.spec

@@ -2,11 +2,15 @@ setup
 {
     -- OCLASS for database changed in PG 16 from 25 to 26
     SELECT CASE WHEN substring(version(), '\d+')::integer < 16 THEN 25 ELSE 26 END AS value INTO oclass_database;
+
+    SELECT 1 FROM citus_add_node('localhost', 57636, 0);
 }
 
 teardown
 {
     DROP TABLE IF EXISTS oclass_database;
+
+    select 1 from citus_remove_node('localhost', 57636);
 }
 
 session "s1"

src/test/regress/sql/alter_database_propagation.sql

@@ -49,6 +49,7 @@ alter database regression set lock_timeout to DEFAULT;
 alter database regression RESET lock_timeout;
 
 set citus.enable_create_database_propagation=on;
+SET citus.next_operation_id TO 3000;
 create database "regression!'2";
 alter database "regression!'2" with CONNECTION LIMIT 100;
 alter database "regression!'2" with IS_TEMPLATE true CONNECTION LIMIT 50;
@@ -90,6 +91,7 @@ set citus.enable_create_database_propagation=on;
 
 drop database regression3;
 
+SET citus.next_operation_id TO 3100;
 create database "regression!'4";
 
 

src/test/regress/sql/create_drop_database_propagation.sql

@@ -218,7 +218,8 @@ SELECT * FROM public.check_database_on_all_nodes('my_template_database') ORDER B
 --tests for special characters in database name
 set citus.enable_create_database_propagation=on;
 SET citus.log_remote_commands = true;
-set citus.grep_remote_commands = '%CREATE DATABASE%';
+set citus.grep_remote_commands = '%DATABASE%';
+SET citus.next_operation_id TO 2000;
 
 create database "mydatabase#1'2";
 
@@ -746,6 +747,63 @@ SELECT 1 FROM run_command_on_all_nodes($$REVOKE ALL ON TABLESPACE pg_default FRO
 DROP DATABASE no_createdb;
 DROP USER no_createdb;
 
+-- Test a failure scenario by trying to create a distributed database that
+-- already exists on one of the nodes.
+
+\c - - - :worker_1_port
+CREATE DATABASE "test_\!failure";
+
+\c - - - :master_port
+
+SET citus.enable_create_database_propagation TO ON;
+
+CREATE DATABASE "test_\!failure";
+
+SET client_min_messages TO WARNING;
+CALL citus_cleanup_orphaned_resources();
+RESET client_min_messages;
+
+SELECT result AS database_cleanedup_on_node FROM run_command_on_all_nodes($$SELECT COUNT(*)=0 FROM pg_database WHERE datname LIKE 'citus_temp_database_%'$$);
+SELECT * FROM public.check_database_on_all_nodes($$test_\!failure$$) ORDER BY node_type, result;
+
+SET citus.enable_create_database_propagation TO OFF;
+CREATE DATABASE "test_\!failure1";
+
+\c - - - :worker_1_port
+DROP DATABASE "test_\!failure";
+
+SET citus.enable_create_database_propagation TO ON;
+
+CREATE DATABASE "test_\!failure1";
+
+SET client_min_messages TO WARNING;
+CALL citus_cleanup_orphaned_resources();
+RESET client_min_messages;
+
+SELECT result AS database_cleanedup_on_node FROM run_command_on_all_nodes($$SELECT COUNT(*)=0 FROM pg_database WHERE datname LIKE 'citus_temp_database_%'$$);
+SELECT * FROM public.check_database_on_all_nodes($$test_\!failure1$$) ORDER BY node_type, result;
+
+\c - - - :master_port
+
+-- Before dropping local "test_\!failure1" database, test a failure scenario
+-- by trying to create a distributed database that already exists "on local
+-- node" this time.
+
+SET citus.enable_create_database_propagation TO ON;
+
+CREATE DATABASE "test_\!failure1";
+
+SET client_min_messages TO WARNING;
+CALL citus_cleanup_orphaned_resources();
+RESET client_min_messages;
+
+SELECT result AS database_cleanedup_on_node FROM run_command_on_all_nodes($$SELECT COUNT(*)=0 FROM pg_database WHERE datname LIKE 'citus_temp_database_%'$$);
+SELECT * FROM public.check_database_on_all_nodes($$test_\!failure1$$) ORDER BY node_type, result;
+
+SET citus.enable_create_database_propagation TO OFF;
+
+DROP DATABASE "test_\!failure1";
+
 SET citus.enable_create_database_propagation TO ON;
 
 --clean up resources created by this test

src/test/regress/sql/failure_create_database.sql

@@ -0,0 +1,128 @@
+SET citus.enable_create_database_propagation TO ON;
+SET client_min_messages TO WARNING;
+
+SELECT 1 FROM citus_add_node('localhost', :master_port, 0);
+
+CREATE FUNCTION get_temp_databases_on_nodes()
+RETURNS TEXT AS $func$
+  SELECT array_agg(DISTINCT result ORDER BY result) AS temp_databases_on_nodes FROM run_command_on_all_nodes($$SELECT datname FROM pg_database WHERE datname LIKE 'citus_temp_database_%'$$) WHERE result != '';
+$func$
+LANGUAGE sql;
+
+CREATE FUNCTION count_db_cleanup_records()
+RETURNS TABLE(object_name TEXT, count INTEGER) AS $func$
+  SELECT object_name, COUNT(*) FROM pg_dist_cleanup WHERE object_name LIKE 'citus_temp_database_%' GROUP BY object_name;
+$func$
+LANGUAGE sql;
+
+CREATE FUNCTION ensure_no_temp_databases_on_any_nodes()
+RETURNS BOOLEAN AS $func$
+  SELECT bool_and(result::boolean) AS no_temp_databases_on_any_nodes FROM run_command_on_all_nodes($$SELECT COUNT(*)=0 FROM pg_database WHERE datname LIKE 'citus_temp_database_%'$$);
+$func$
+LANGUAGE sql;
+
+-- cleanup any orphaned resources from previous runs
+CALL citus_cleanup_orphaned_resources();
+
+SET citus.next_operation_id TO 4000;
+
+ALTER SYSTEM SET citus.defer_shard_delete_interval TO -1;
+SELECT pg_reload_conf();
+SELECT pg_sleep(0.1);
+
+SELECT citus.mitmproxy('conn.kill()');
+CREATE DATABASE db1;
+SELECT citus.mitmproxy('conn.allow()');
+
+SELECT get_temp_databases_on_nodes();
+SELECT * FROM count_db_cleanup_records();
+CALL citus_cleanup_orphaned_resources();
+SELECT ensure_no_temp_databases_on_any_nodes();
+SELECT * FROM public.check_database_on_all_nodes($$db1$$) ORDER BY node_type, result;
+
+SELECT citus.mitmproxy('conn.onQuery(query="^CREATE DATABASE").cancel(' || pg_backend_pid() || ')');
+CREATE DATABASE db1;
+SELECT citus.mitmproxy('conn.allow()');
+
+SELECT get_temp_databases_on_nodes();
+SELECT * FROM count_db_cleanup_records();
+CALL citus_cleanup_orphaned_resources();
+SELECT ensure_no_temp_databases_on_any_nodes();
+SELECT * FROM public.check_database_on_all_nodes($$db1$$) ORDER BY node_type, result;
+
+SELECT citus.mitmproxy('conn.onQuery(query="^ALTER DATABASE").cancel(' || pg_backend_pid() || ')');
+CREATE DATABASE db1;
+SELECT citus.mitmproxy('conn.allow()');
+
+SELECT get_temp_databases_on_nodes();
+SELECT * FROM count_db_cleanup_records();
+CALL citus_cleanup_orphaned_resources();
+SELECT ensure_no_temp_databases_on_any_nodes();
+SELECT * FROM public.check_database_on_all_nodes($$db1$$) ORDER BY node_type, result;
+
+SELECT citus.mitmproxy('conn.onQuery(query="^BEGIN TRANSACTION ISOLATION LEVEL READ COMMITTED").kill()');
+CREATE DATABASE db1;
+SELECT citus.mitmproxy('conn.allow()');
+
+SELECT get_temp_databases_on_nodes();
+SELECT * FROM count_db_cleanup_records();
+CALL citus_cleanup_orphaned_resources();
+SELECT ensure_no_temp_databases_on_any_nodes();
+SELECT * FROM public.check_database_on_all_nodes($$db1$$) ORDER BY node_type, result;
+
+SELECT citus.mitmproxy('conn.onQuery(query="^PREPARE TRANSACTION").kill()');
+CREATE DATABASE db1;
+SELECT citus.mitmproxy('conn.allow()');
+
+SELECT get_temp_databases_on_nodes();
+SELECT * FROM count_db_cleanup_records();
+CALL citus_cleanup_orphaned_resources();
+SELECT ensure_no_temp_databases_on_any_nodes();
+SELECT * FROM public.check_database_on_all_nodes($$db1$$) ORDER BY node_type, result;
+
+SELECT citus.mitmproxy('conn.onQuery(query="^COMMIT PREPARED").kill()');
+CREATE DATABASE db1;
+SELECT citus.mitmproxy('conn.allow()');
+
+-- not call citus_cleanup_orphaned_resources() but recover the prepared transactions this time
+SELECT 1 FROM recover_prepared_transactions();
+SELECT ensure_no_temp_databases_on_any_nodes();
+SELECT * FROM public.check_database_on_all_nodes($$db1$$) ORDER BY node_type, result;
+
+DROP DATABASE db1;
+
+-- after recovering the prepared transactions, cleanup records should also be removed
+SELECT * FROM count_db_cleanup_records();
+
+SELECT citus.mitmproxy('conn.onQuery(query="^SELECT citus_internal.acquire_citus_advisory_object_class_lock").kill()');
+CREATE DATABASE db1;
+SELECT citus.mitmproxy('conn.allow()');
+
+SELECT get_temp_databases_on_nodes();
+SELECT * FROM count_db_cleanup_records();
+CALL citus_cleanup_orphaned_resources();
+SELECT ensure_no_temp_databases_on_any_nodes();
+SELECT * FROM public.check_database_on_all_nodes($$db1$$) ORDER BY node_type, result;
+
+SELECT citus.mitmproxy('conn.onParse(query="^WITH distributed_object_data").kill()');
+CREATE DATABASE db1;
+SELECT citus.mitmproxy('conn.allow()');
+
+SELECT get_temp_databases_on_nodes();
+SELECT * FROM count_db_cleanup_records();
+CALL citus_cleanup_orphaned_resources();
+SELECT ensure_no_temp_databases_on_any_nodes();
+SELECT * FROM public.check_database_on_all_nodes($$db1$$) ORDER BY node_type, result;
+
+CREATE DATABASE db1;
+
+-- show that a successful database creation doesn't leave any pg_dist_cleanup records behind
+SELECT * FROM count_db_cleanup_records();
+
+DROP DATABASE db1;
+
+DROP FUNCTION get_temp_databases_on_nodes();
+DROP FUNCTION ensure_no_temp_databases_on_any_nodes();
+DROP FUNCTION count_db_cleanup_records();
+
+SELECT 1 FROM citus_remove_node('localhost', :master_port);