diff --git a/src/backend/distributed/commands/database.c b/src/backend/distributed/commands/database.c
index 33223f41686..5479a59edcf 100644
--- a/src/backend/distributed/commands/database.c
+++ b/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.
 		 */
-
-		return NontransactionalNodeDDLTaskList(NON_COORDINATOR_NODES, commands);
+		bool warnForPartialFailure = true;
+		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
- * all workers to create the database.
+ * postgres instance.
  *
+ * 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);
-	return createDatabaseDDLJobList;
+		NontransactionalNodeDDLTaskList(REMOTE_NODES, createDatabaseCommands,
+										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;
 }
 
diff --git a/src/backend/distributed/commands/index.c b/src/backend/distributed/commands/index.c
index c4113617666..e97312df271 100644
--- a/src/backend/distributed/commands/index.c
+++ b/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);
 	}
diff --git a/src/backend/distributed/commands/utility_hook.c b/src/backend/distributed/commands/utility_hook.c
index a1a23331076..7dff9cbf6cb 100644
--- a/src/backend/distributed/commands/utility_hook.c
+++ b/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;
 }
diff --git a/src/backend/distributed/operations/shard_cleaner.c b/src/backend/distributed/operations/shard_cleaner.c
index db1cad6bcae..2efce9a7b09 100644
--- a/src/backend/distributed/operations/shard_cleaner.c
+++ b/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.
diff --git a/src/include/distributed/commands/utility_hook.h b/src/include/distributed/commands/utility_hook.h
index 9046c73093e..52fcf70912c 100644
--- a/src/include/distributed/commands/utility_hook.h
+++ b/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);
diff --git a/src/include/distributed/shard_cleaner.h b/src/include/distributed/shard_cleaner.h
index 4967846b2ba..7609bd90024 100644
--- a/src/include/distributed/shard_cleaner.h
+++ b/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;
 
 /*
diff --git a/src/test/regress/expected/alter_database_propagation.out b/src/test/regress/expected/alter_database_propagation.out
index f01d39ab911..5c45a25e29c 100644
--- a/src/test/regress/expected/alter_database_propagation.out
+++ b/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"
diff --git a/src/test/regress/expected/create_drop_database_propagation.out b/src/test/regress/expected/create_drop_database_propagation.out
index da4ec4eb711..4ddbaae3fe3 100644
--- a/src/test/regress/expected/create_drop_database_propagation.out
+++ b/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.
diff --git a/src/test/regress/expected/failure_create_database.out b/src/test/regress/expected/failure_create_database.out
new file mode 100644
index 00000000000..81fcd451965
--- /dev/null
+++ b/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)
+
diff --git a/src/test/regress/expected/isolation_database_cmd_from_any_node.out b/src/test/regress/expected/isolation_database_cmd_from_any_node.out
index 771c67fe88f..e952bb45728 100644
--- a/src/test/regress/expected/isolation_database_cmd_from_any_node.out
+++ b/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)
+
diff --git a/src/test/regress/failure_schedule b/src/test/regress/failure_schedule
index e1ad362b518..8b992422ef2 100644
--- a/src/test/regress/failure_schedule
+++ b/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
diff --git a/src/test/regress/spec/isolation_database_cmd_from_any_node.spec b/src/test/regress/spec/isolation_database_cmd_from_any_node.spec
index 1e004cb338f..8637a8942b6 100644
--- a/src/test/regress/spec/isolation_database_cmd_from_any_node.spec
+++ b/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"
diff --git a/src/test/regress/sql/alter_database_propagation.sql b/src/test/regress/sql/alter_database_propagation.sql
index 4904919a6a6..9a8b1fab8af 100644
--- a/src/test/regress/sql/alter_database_propagation.sql
+++ b/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";
 
 
diff --git a/src/test/regress/sql/create_drop_database_propagation.sql b/src/test/regress/sql/create_drop_database_propagation.sql
index 329f4861203..de55258c3c5 100644
--- a/src/test/regress/sql/create_drop_database_propagation.sql
+++ b/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
diff --git a/src/test/regress/sql/failure_create_database.sql b/src/test/regress/sql/failure_create_database.sql
new file mode 100644
index 00000000000..d117dc81192
--- /dev/null
+++ b/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);