external
/
citus
mirror of https://github.com/citusdata/citus.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Lock parent shard resource while creating new partitions

fix/distributed_deadlock1
Sait Talha Nisanci 2021-08-13 16:49:08 +03:00
parent 364f902ff0
commit d3faf974ad
2 changed files with 36 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

34
src/backend/distributed/executor/adaptive_executor.c
View File

@ -704,6 +704,7 @@ static void SetAttributeInputMetadata(DistributedExecution *execution,
static void LookupTaskPlacementHostAndPort(ShardPlacement *taskPlacement, char **nodeName, static void LookupTaskPlacementHostAndPort(ShardPlacement *taskPlacement, char **nodeName,
int *nodePort); int *nodePort);
static bool IsDummyPlacement(ShardPlacement *taskPlacement); static bool IsDummyPlacement(ShardPlacement *taskPlacement);
static void LockParentShardResouceIfPartitionTaskList(List *taskList);
/* /*
* AdaptiveExecutorPreExecutorRun gets called right before postgres starts its executor * AdaptiveExecutorPreExecutorRun gets called right before postgres starts its executor
@ -1609,6 +1610,11 @@ AcquireExecutorShardLocksForExecution(DistributedExecution *execution)
/* acquire the locks for both the remote and local tasks */ /* acquire the locks for both the remote and local tasks */
List *taskList = execution->remoteAndLocalTaskList; List *taskList = execution->remoteAndLocalTaskList;
if (modLevel == ROW_MODIFY_NONE)
{
LockParentShardResouceIfPartitionTaskList(taskList);
}
if (modLevel <= ROW_MODIFY_READONLY && if (modLevel <= ROW_MODIFY_READONLY &&
!SelectForUpdateOnReferenceTable(taskList)) !SelectForUpdateOnReferenceTable(taskList))
{ {
@ -1638,6 +1644,34 @@ AcquireExecutorShardLocksForExecution(DistributedExecution *execution)
} }
/*
* LockParentShardResouceIfPartitionTaskList locks the parent shard
* resource if the given taskList is on a partition table.
*/
static void
LockParentShardResouceIfPartitionTaskList(List *taskList)
{
if (list_length(taskList) < 1)
{
return;
}
Task *task = (Task *) linitial(taskList);
uint64 shardId = task->anchorShardId;
if (shardId == INVALID_SHARD_ID)
{
return;
}
ShardInterval *shardInterval = LoadShardInterval(shardId);
Oid relationId = shardInterval->relationId;
if (PartitionTable(relationId))
{
LockParentShardResourceIfPartition(shardId, AccessExclusiveLock);
}
}
/* /*
* FinishDistributedExecution cleans up resources associated with a * FinishDistributedExecution cleans up resources associated with a
* distributed execution. * distributed execution.