commit

Author: onurctirtir

src/backend/distributed/commands/database.c

@@ -41,6 +41,7 @@
 #include "distributed/metadata_utility.h"
 #include "distributed/multi_executor.h"
 #include "distributed/relation_access_tracking.h"
+#include "distributed/serialize_distributed_ddls.h"
 #include "distributed/worker_protocol.h"
 #include "distributed/worker_transaction.h"
 
@@ -161,6 +162,10 @@ get_database_owner(Oid dbId)
  *
  * In this stage we can prepare the commands that need to be run on all workers to grant
  * on databases.
+ *
+ * We also serialize database commands globally by acquiring Citus specific advisory
+ * locks on affected databases based on OCLASS_DATABASE on the first primary worker
+ * node.
  */
 List *
 PreprocessGrantOnDatabaseStmt(Node *node, const char *queryString,
@@ -183,6 +188,24 @@ PreprocessGrantOnDatabaseStmt(Node *node, const char *queryString,
 
 	EnsureCoordinator();
 
+	/*
+	 * Sort before acquring the locks to prevent deadlocks that could happen
+	 * due to acquiring the locks in different orders.
+	 */
+	List *sortedDistributedDatabases = SortList(distributedDatabases, CompareStringNodes);
+
+	String *distributedDatabaseName = NULL;
+	foreach_ptr(distributedDatabaseName, sortedDistributedDatabases)
+	{
+		/* FilterDistributedDatabases ensured that the database exists and is distributed */
+		bool missingOk = false;
+		ObjectAddress *distributedDatabaseAddress = GetDatabaseAddressFromDatabaseName(
+			strVal(distributedDatabaseName), missingOk);
+
+		SerializeDistributedDDLsOnObjectClass(OCLASS_DATABASE,
+											  distributedDatabaseAddress->objectId);
+	}
+
 	List *originalObjects = stmt->objects;
 
 	stmt->objects = distributedDatabases;
@@ -248,6 +271,9 @@ IsSetTablespaceStatement(AlterDatabaseStmt *stmt)
  *
  * In this stage we can prepare the commands that need to be run on all workers to grant
  * on databases.
+ *
+ * We also serialize database commands globally by acquiring a Citus specific advisory
+ * lock based on OCLASS_DATABASE on the first primary worker node.
  */
 List *
 PreprocessAlterDatabaseStmt(Node *node, const char *queryString,
@@ -264,6 +290,7 @@ PreprocessAlterDatabaseStmt(Node *node, const char *queryString,
 	}
 
 	EnsureCoordinator();
+	SerializeDistributedDDLsOnObjectClass(OCLASS_DATABASE, dbAddress->objectId);
 
 	char *sql = DeparseTreeNode((Node *) stmt);
 
@@ -296,6 +323,9 @@ PreprocessAlterDatabaseStmt(Node *node, const char *queryString,
  *
  * In this stage we can prepare the commands that need to be run on all workers to grant
  * on databases.
+ *
+ * We also serialize database commands globally by acquiring a Citus specific advisory
+ * lock based on OCLASS_DATABASE on the first primary worker node.
  */
 List *
 PreprocessAlterDatabaseRefreshCollStmt(Node *node, const char *queryString,
@@ -312,6 +342,7 @@ PreprocessAlterDatabaseRefreshCollStmt(Node *node, const char *queryString,
 	}
 
 	EnsureCoordinator();
+	SerializeDistributedDDLsOnObjectClass(OCLASS_DATABASE, dbAddress->objectId);
 
 	char *sql = DeparseTreeNode((Node *) stmt);
 
@@ -325,8 +356,51 @@ PreprocessAlterDatabaseRefreshCollStmt(Node *node, const char *queryString,
 
 #endif
 
+
+/*
+ * PreprocessAlterDatabaseRenameStmt is executed before the statement is applied to
+ * the local postgres instance.
+ *
+ * We also serialize database commands globally by acquiring a Citus specific advisory
+ * lock based on OCLASS_DATABASE on the first primary worker node.
+ *
+ * We acquire this lock here instead of PostprocessAlterDatabaseRenameStmt because the
+ * command renames the database and SerializeDistributedDDLsOnObjectClass resolves the
+ * object on workers based on database name. For this reason, we need to acquire the lock
+ * before the command is applied to the local postgres instance.
+ */
+List *
+PreprocessAlterDatabaseRenameStmt(Node *node, const char *queryString,
+								  ProcessUtilityContext processUtilityContext)
+{
+	bool missingOk = true;
+	RenameStmt *stmt = castNode(RenameStmt, node);
+	ObjectAddress *dbAddress = GetDatabaseAddressFromDatabaseName(stmt->subname,
+																  missingOk);
+
+	if (!ShouldPropagate() || !IsAnyObjectDistributed(list_make1(dbAddress)))
+	{
+		return NIL;
+	}
+
+	EnsureCoordinator();
+
+	/*
+	 * Different than other ALTER DATABASE commands, we first acquire a lock
+	 * by providing InvalidOid because we want ALTER TABLE .. RENAME TO ..
+	 * commands to block not only with ALTER DATABASE operations but also
+	 * with CREATE DATABASE operations because they might cause name conflicts
+	 * and that could also cause deadlocks too.
+	 */
+	SerializeDistributedDDLsOnObjectClass(OCLASS_DATABASE, InvalidOid);
+	SerializeDistributedDDLsOnObjectClass(OCLASS_DATABASE, dbAddress->objectId);
+
+	return NIL;
+}
+
+
 /*
- * PreprocessAlterDatabaseRenameStmt is executed before the statement is applied to the local
+ * PostprocessAlterDatabaseRenameStmt is executed after the statement is applied to the local
  * postgres instance. In this stage we prepare ALTER DATABASE RENAME statement to be run on
  * all workers.
  */
@@ -361,6 +435,9 @@ PostprocessAlterDatabaseRenameStmt(Node *node, const char *queryString)
  *
  * In this stage we can prepare the commands that need to be run on all workers to grant
  * on databases.
+ *
+ * We also serialize database commands globally by acquiring a Citus specific advisory
+ * lock based on OCLASS_DATABASE on the first primary worker node.
  */
 List *
 PreprocessAlterDatabaseSetStmt(Node *node, const char *queryString,
@@ -377,6 +454,7 @@ PreprocessAlterDatabaseSetStmt(Node *node, const char *queryString,
 	}
 
 	EnsureCoordinator();
+	SerializeDistributedDDLsOnObjectClass(OCLASS_DATABASE, dbAddress->objectId);
 
 	char *sql = DeparseTreeNode((Node *) stmt);
 
@@ -395,6 +473,9 @@ 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.
+ *
+ * We also serialize database commands globally by acquiring a Citus specific advisory
+ * lock based on OCLASS_DATABASE on the first primary worker node.
  */
 List *
 PreprocessCreateDatabaseStmt(Node *node, const char *queryString,
@@ -405,11 +486,13 @@ PreprocessCreateDatabaseStmt(Node *node, const char *queryString,
 		return NIL;
 	}
 
-	EnsureCoordinator();
+	EnsurePropagationToCoordinator();
 
 	CreatedbStmt *stmt = castNode(CreatedbStmt, node);
 	EnsureSupportedCreateDatabaseCommand(stmt);
 
+	SerializeDistributedDDLsOnObjectClass(OCLASS_DATABASE, InvalidOid);
+
 	return NIL;
 }
 
@@ -430,7 +513,7 @@ PostprocessCreateDatabaseStmt(Node *node, const char *queryString)
 		return NIL;
 	}
 
-	EnsureCoordinator();
+	EnsurePropagationToCoordinator();
 
 	/*
 	 * Given that CREATE DATABASE doesn't support "IF NOT EXISTS" and we're
@@ -448,7 +531,7 @@ PostprocessCreateDatabaseStmt(Node *node, const char *queryString)
 								(void *) createDatabaseCommand,
 								ENABLE_DDL_PROPAGATION);
 
-	return NontransactionalNodeDDLTaskList(NON_COORDINATOR_NODES, commands);
+	return NontransactionalNodeDDLTaskList(REMOTE_NODES, commands);
 }
 
 
@@ -458,6 +541,9 @@ PostprocessCreateDatabaseStmt(Node *node, const char *queryString)
  * all workers to drop the database. Since the DROP DATABASE statement gives error in
  * transaction context, we need to use NontransactionalNodeDDLTaskList to send the
  * DROP DATABASE statement to the workers.
+ *
+ * We also serialize database commands globally by acquiring a Citus specific advisory
+ * lock based on OCLASS_DATABASE on the first primary worker node.
  */
 List *
 PreprocessDropDatabaseStmt(Node *node, const char *queryString,
@@ -468,7 +554,7 @@ PreprocessDropDatabaseStmt(Node *node, const char *queryString,
 		return NIL;
 	}
 
-	EnsureCoordinator();
+	EnsurePropagationToCoordinator();
 
 	DropdbStmt *stmt = (DropdbStmt *) node;
 
@@ -488,13 +574,15 @@ PreprocessDropDatabaseStmt(Node *node, const char *queryString,
 		return NIL;
 	}
 
+	SerializeDistributedDDLsOnObjectClass(OCLASS_DATABASE, address->objectId);
+
 	char *dropDatabaseCommand = DeparseTreeNode(node);
 
 	List *commands = list_make3(DISABLE_DDL_PROPAGATION,
 								(void *) dropDatabaseCommand,
 								ENABLE_DDL_PROPAGATION);
 
-	return NontransactionalNodeDDLTaskList(NON_COORDINATOR_NODES, commands);
+	return NontransactionalNodeDDLTaskList(REMOTE_NODES, commands);
 }
 
 

src/backend/distributed/commands/distribute_object_ops.c

@@ -536,7 +536,7 @@ static DistributeObjectOps Database_Set = {
 static DistributeObjectOps Database_Rename = {
 	.deparse = DeparseAlterDatabaseRenameStmt,
 	.qualify = NULL,
-	.preprocess = NULL,
+	.preprocess = PreprocessAlterDatabaseRenameStmt,
 	.postprocess = PostprocessAlterDatabaseRenameStmt,
 	.objectType = OBJECT_DATABASE,
 	.operationType = DIST_OPS_ALTER,

src/backend/distributed/commands/utility_hook.c

@@ -726,9 +726,9 @@ citus_ProcessUtilityInternal(PlannedStmt *pstmt,
 			ereport(NOTICE, (errmsg("Citus partially supports CREATE DATABASE for "
 									"distributed databases"),
 							 errdetail("Citus does not propagate CREATE DATABASE "
-									   "command to workers"),
+									   "command to other nodes"),
 							 errhint("You can manually create a database and its "
-									 "extensions on workers.")));
+									 "extensions on other nodes.")));
 		}
 	}
 	else if (IsA(parsetree, CreateRoleStmt) && !EnableCreateRolePropagation)

src/backend/distributed/metadata/node_metadata.c

@@ -2771,12 +2771,24 @@ EnsureCoordinatorIsInMetadata(void)
 {
 	bool isCoordinatorInMetadata = false;
 	PrimaryNodeForGroup(COORDINATOR_GROUP_ID, &isCoordinatorInMetadata);
-	if (!isCoordinatorInMetadata)
+	if (isCoordinatorInMetadata)
+	{
+		return;
+	}
+
+	/* be more descriptive when we're not on coordinator */
+	if (IsCoordinator())
 	{
 		ereport(ERROR, (errmsg("coordinator is not added to the metadata"),
 						errhint("Use SELECT citus_set_coordinator_host('<hostname>') "
 								"to configure the coordinator hostname")));
 	}
+	else
+	{
+		ereport(ERROR, (errmsg("coordinator is not added to the metadata"),
+						errhint("Use SELECT citus_set_coordinator_host('<hostname>') "
+								"on coordinator to configure the coordinator hostname")));
+	}
 }