Support concurrent partition detach

src/backend/distributed/commands/table.c

@@ -51,6 +51,7 @@
 #include "distributed/multi_partitioning_utils.h"
 #include "distributed/reference_table_utils.h"
 #include "distributed/relation_access_tracking.h"
+#include "distributed/relay_utility.h"
 #include "distributed/resource_lock.h"
 #include "distributed/tenant_schema_metadata.h"
 #include "distributed/version_compat.h"
@@ -115,6 +116,8 @@ static void ErrorIfUnsupportedAlterAddConstraintStmt(AlterTableStmt *alterTableS
 static List * CreateRightShardListForInterShardDDLTask(Oid rightRelationId,
 													   Oid leftRelationId,
 													   List *leftShardList);
+static List * ConcurrentDetachPartitionTaskList(Oid parentRelationId,
+												 Oid partitionRelationId);
 static void SetInterShardDDLTaskPlacementList(Task *task,
 											  ShardInterval *leftShardInterval,
 											  ShardInterval *rightShardInterval);
@@ -1336,6 +1339,7 @@ PreprocessAlterTableStmt(Node *node, const char *alterTableCommand,
 	 */
 	bool deparseAT = false;
 	bool propagateCommandToRemoteNodes = true;
+	bool concurrentPartitionDetach = false;
 
 	/*
 	 * Sometimes we want to run a different DDL Command string on remote MX workers
@@ -1666,6 +1670,18 @@ PreprocessAlterTableStmt(Node *node, const char *alterTableCommand,
 			Assert(list_length(commandList) <= 1);
 
 			rightRelationId = RangeVarGetRelid(partitionCommand->name, NoLock, false);
+			concurrentPartitionDetach = partitionCommand->concurrent;
+		}
+		else if (alterTableType == AT_DetachPartitionFinalize)
+		{
+			PartitionCmd *partitionCommand = (PartitionCmd *) command->def;
+
+			rightRelationId = RangeVarGetRelid(partitionCommand->name, NoLock, false);
+			concurrentPartitionDetach = true;
+			alterTableCommand = psprintf("ALTER TABLE %s DETACH PARTITION %s "
+										 "CONCURRENTLY",
+										 generate_qualified_relation_name(leftRelationId),
+										 generate_qualified_relation_name(rightRelationId));
 		}
 		else if (AlterTableCommandTypeIsTrigger(alterTableType))
 		{
@@ -1709,8 +1725,17 @@ PreprocessAlterTableStmt(Node *node, const char *alterTableCommand,
 		else
 		{
 			/* if foreign key or attaching partition index related, use specialized task list function ... */
-			ddlJob->taskList = InterShardDDLTaskList(leftRelationId, rightRelationId,
+			if (concurrentPartitionDetach)
+			{
+				ddlJob->taskList = ConcurrentDetachPartitionTaskList(leftRelationId,
+															 rightRelationId);
+				ddlJob->warnForPartialFailure = true;
+			}
+			else
+			{
+				ddlJob->taskList = InterShardDDLTaskList(leftRelationId, rightRelationId,
 													 alterTableCommand);
+			}
 		}
 	}
 	else
@@ -3813,8 +3838,6 @@ ErrorIfUnsupportedAlterTableStmt(AlterTableStmt *alterTableStatement)
 
 			case AT_DetachPartitionFinalize:
 			{
-				ereport(ERROR, (errmsg("ALTER TABLE .. DETACH PARTITION .. FINALIZE "
-									   "commands are currently unsupported.")));
 				break;
 			}
 
@@ -3830,15 +3853,6 @@ ErrorIfUnsupportedAlterTableStmt(AlterTableStmt *alterTableStatement)
 											"separately.")));
 				}
 
-				PartitionCmd *partitionCommand = (PartitionCmd *) command->def;
-
-				if (partitionCommand->concurrent)
-				{
-					ereport(ERROR, (errmsg("ALTER TABLE .. DETACH PARTITION .. "
-										   "CONCURRENTLY commands are currently "
-										   "unsupported.")));
-				}
-
 				break;
 			}
 
@@ -4152,6 +4166,60 @@ InterShardDDLTaskList(Oid leftRelationId, Oid rightRelationId,
 }
 
 
+/*
+ * Build top-level commands for DETACH PARTITION CONCURRENTLY.  Unlike other
+ * inter-shard DDL, these commands cannot run through
+ * worker_apply_inter_shard_ddl_command(), because PostgreSQL implements the
+ * detach using multiple transactions.
+ */
+static List *
+ConcurrentDetachPartitionTaskList(Oid parentRelationId, Oid partitionRelationId)
+{
+	List *parentShardList = LoadShardIntervalList(parentRelationId);
+	List *partitionShardList = CreateRightShardListForInterShardDDLTask(
+		partitionRelationId, parentRelationId, parentShardList);
+	List *taskList = NIL;
+	uint64 jobId = INVALID_JOB_ID;
+	int taskId = 1;
+	char *parentSchemaName = get_namespace_name(get_rel_namespace(parentRelationId));
+	char *partitionSchemaName = get_namespace_name(get_rel_namespace(partitionRelationId));
+	char *parentRelationName = get_rel_name(parentRelationId);
+	char *partitionRelationName = get_rel_name(partitionRelationId);
+
+	LockShardListMetadata(parentShardList, ShareLock);
+
+	ShardInterval *parentShard = NULL;
+	ShardInterval *partitionShard = NULL;
+	forboth_ptr(parentShard, parentShardList, partitionShard, partitionShardList)
+	{
+		char *parentShardName = pstrdup(parentRelationName);
+		char *partitionShardName = pstrdup(partitionRelationName);
+		Task *task = CitusMakeNode(Task);
+
+		AppendShardIdToName(&parentShardName, parentShard->shardId);
+		AppendShardIdToName(&partitionShardName, partitionShard->shardId);
+
+		task->jobId = jobId;
+		task->taskId = taskId++;
+		task->taskType = DDL_TASK;
+		SetTaskQueryString(task,
+			psprintf("ALTER TABLE %s DETACH PARTITION %s CONCURRENTLY",
+					 quote_qualified_identifier(parentSchemaName, parentShardName),
+					 quote_qualified_identifier(partitionSchemaName,
+											partitionShardName)));
+		task->replicationModel = REPLICATION_MODEL_INVALID;
+		task->anchorShardId = parentShard->shardId;
+		task->cannotBeExecutedInTransaction = true;
+		SetInterShardDDLTaskPlacementList(task, parentShard, partitionShard);
+		SetInterShardDDLTaskRelationShardList(task, parentShard, partitionShard);
+
+		taskList = lappend(taskList, task);
+	}
+
+	return taskList;
+}
+
+
 /*
  * CreateRightShardListForInterShardDDLTask is a helper function that creates
  * shard list for the right relation for InterShardDDLTaskList.

src/test/regress/expected/detach_partition_concurrently.out

@@ -0,0 +1,45 @@
+CREATE SCHEMA detach_partition_concurrently;
+SET search_path TO detach_partition_concurrently;
+SET citus.shard_count TO 2;
+SET citus.shard_replication_factor TO 1;
+CREATE TABLE parent (a int) PARTITION BY RANGE (a);
+CREATE TABLE child PARTITION OF parent FOR VALUES FROM (0) TO (10);
+SELECT create_distributed_table('parent', 'a');
+ create_distributed_table
+---------------------------------------------------------------------
+
+(1 row)
+
+ALTER TABLE parent DETACH PARTITION child CONCURRENTLY;
+SELECT relispartition
+FROM pg_class
+WHERE oid = 'child'::regclass;
+ relispartition
+---------------------------------------------------------------------
+ f
+(1 row)
+
+SELECT result
+FROM run_command_on_workers($$
+	SELECT count(*)
+	FROM pg_inherits i
+	JOIN pg_class p ON p.oid = i.inhparent
+	WHERE p.relname LIKE 'parent\_%'
+$$)
+ORDER BY result;
+ result
+---------------------------------------------------------------------
+ 0
+ 0
+(2 rows)
+
+INSERT INTO child VALUES (1);
+SELECT * FROM child;
+ a
+---------------------------------------------------------------------
+ 1
+(1 row)
+
+DROP TABLE child;
+DROP TABLE parent;
+DROP SCHEMA detach_partition_concurrently;

src/test/regress/expected/isolation_detach_partition_concurrently.out

@@ -0,0 +1,87 @@
+Parsed test spec with 3 sessions
+
+starting permutation: s1_begin s1_read s2_detach s3_cancel s3_pending s3_workers_attached s1_commit s3_finalize s3_done s3_workers_done
+create_distributed_table
+---------------------------------------------------------------------
+
+(1 row)
+
+step s1_begin: BEGIN;
+step s1_read: SELECT * FROM detach_parent;
+a
+-
+(0 rows)
+
+step s2_detach:
+	ALTER TABLE detach_parent DETACH PARTITION detach_child CONCURRENTLY;
+ <waiting ...>
+step s3_cancel: 
+	SELECT pg_cancel_backend(pid)
+	FROM pg_stat_activity
+	WHERE pid <> pg_backend_pid()
+	  AND query LIKE
+		'%ALTER TABLE detach_parent DETACH PARTITION detach_child CONCURRENTLY%';
+
+pg_cancel_backend
+---------------------------------------------------------------------
+t
+(1 row)
+
+step s2_detach: <... completed>
+ERROR:  canceling statement due to user request
+step s3_pending:
+	SELECT inhdetachpending
+	FROM pg_inherits
+	WHERE inhparent = 'detach_parent'::regclass;
+
+inhdetachpending
+---------------------------------------------------------------------
+t
+(1 row)
+
+step s3_workers_attached:
+	SELECT result
+	FROM run_command_on_workers($$
+		SELECT count(*)
+		FROM pg_inherits i
+		JOIN pg_class p ON p.oid = i.inhparent
+		WHERE p.relname LIKE 'detach_parent\_%'
+	$$)
+	ORDER BY result;
+
+result
+---------------------------------------------------------------------
+     1
+     1
+(2 rows)
+
+step s1_commit: COMMIT;
+step s3_finalize:
+	ALTER TABLE detach_parent DETACH PARTITION detach_child FINALIZE;
+
+step s3_done:
+	SELECT relispartition
+	FROM pg_class
+	WHERE oid = 'detach_child'::regclass;
+
+relispartition
+---------------------------------------------------------------------
+f
+(1 row)
+
+step s3_workers_done:
+	SELECT result
+	FROM run_command_on_workers($$
+		SELECT count(*)
+		FROM pg_inherits i
+		JOIN pg_class p ON p.oid = i.inhparent
+		WHERE p.relname LIKE 'detach_parent\_%'
+	$$)
+	ORDER BY result;
+
+result
+---------------------------------------------------------------------
+     0
+     0
+(2 rows)
+

src/test/regress/expected/pg14.out

@@ -220,24 +220,24 @@ CREATE STATISTICS s1 (dependencies) ON a, b FROM tbl1;
 CREATE STATISTICS s2 (mcv) ON a, b FROM tbl1;
 CREATE STATISTICS s3 (ndistinct) ON date_trunc('month', a), date_trunc('day', a) FROM tbl1;
 set citus.log_remote_commands to off;
--- error out in case of ALTER TABLE .. DETACH PARTITION .. CONCURRENTLY/FINALIZE
--- only if it's a distributed partitioned table
+-- concurrent detach works for distributed partitioned tables
 CREATE TABLE par (a INT UNIQUE) PARTITION BY RANGE(a);
 CREATE TABLE par_1 PARTITION OF par FOR VALUES FROM (1) TO (4);
 CREATE TABLE par_2 PARTITION OF par FOR VALUES FROM (5) TO (8);
--- works as it's not distributed
 ALTER TABLE par DETACH PARTITION par_1 CONCURRENTLY;
--- errors out
 SELECT create_distributed_table('par','a');
  create_distributed_table
 ---------------------------------------------------------------------
 
 (1 row)
 
 ALTER TABLE par DETACH PARTITION par_2 CONCURRENTLY;
-ERROR:  ALTER TABLE .. DETACH PARTITION .. CONCURRENTLY commands are currently unsupported.
-ALTER TABLE par DETACH PARTITION par_2 FINALIZE;
-ERROR:  ALTER TABLE .. DETACH PARTITION .. FINALIZE commands are currently unsupported.
+SELECT relispartition FROM pg_class WHERE oid = 'par_2'::regclass;
+ relispartition
+---------------------------------------------------------------------
+ f
+(1 row)
+
 -- test column compression propagation in distribution
 SET citus.shard_replication_factor TO 1;
 CREATE TABLE col_compression (a TEXT COMPRESSION pglz, b TEXT);

src/test/regress/isolation_schedule

@@ -115,6 +115,9 @@ test: isolation_concurrent_move_create_table
 test: isolation_merge
 test: isolation_merge_replicated
 
+# Concurrent partition detach and interrupted-operation recovery
+test: isolation_detach_partition_concurrently
+
 
 # Note: Always keep this test at the end
 test: isolation_check_mx

src/test/regress/multi_schedule

@@ -41,7 +41,7 @@ test: ensure_no_intermediate_data_leak
 # ----------
 # Tests for partitioning support
 # ----------
-test: multi_partitioning_utils multi_partitioning replicated_partitioned_table
+test: multi_partitioning_utils multi_partitioning replicated_partitioned_table detach_partition_concurrently
 
 
 # ----------