database/doris
2
0
Fork 0
Code Issues Pull Requests Actions 3 Packages Projects Releases Wiki Activity

[fix](race) fix access colocate group ids race #36444 (#36501)

Browse Source 
cherry pick from #36444
This commit is contained in:
yujun
2024-06-19 15:38:50 +08:00
committed by GitHub
parent 349b943e12
commit 9c896efe0b
2 changed files with 17 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
fe/fe-core/src/main/java/org/apache/doris/catalog/ColocateTableIndex.java
View File

@ -429,7 +429,7 @@ public class ColocateTableIndex implements Writable {
public Set<GroupId> getAllGroupIds() {
readLock();
try {
return group2Tables.keySet();
return Sets.newHashSet(group2Tables.keySet());
} finally {
readUnlock();
}

18
fe/fe-core/src/main/java/org/apache/doris/clone/ColocateTableCheckerAndBalancer.java
View File

@ -378,7 +378,7 @@ public class ColocateTableCheckerAndBalancer extends MasterDaemon {
* A B C D
*/
private void relocateAndBalanceGroups() {
Set<GroupId> groupIds = Sets.newHashSet(Env.getCurrentEnv().getColocateTableIndex().getAllGroupIds());
Set<GroupId> groupIds = Env.getCurrentEnv().getColocateTableIndex().getAllGroupIds();
// balance only inside each group, excluded balance between all groups
Set<GroupId> changeGroups = relocateAndBalanceGroup(groupIds, false);
@ -410,6 +410,10 @@ public class ColocateTableCheckerAndBalancer extends MasterDaemon {
}
ColocateGroupSchema groupSchema = colocateIndex.getGroupSchema(groupId);
if (groupSchema == null) {
LOG.info("Not found colocate group {}, maybe delete", groupId);
continue;
}
ReplicaAllocation replicaAlloc = groupSchema.getReplicaAlloc();
try {
Env.getCurrentSystemInfo().checkReplicaAllocation(replicaAlloc);
@ -480,13 +484,18 @@ public class ColocateTableCheckerAndBalancer extends MasterDaemon {
// check each group
Set<GroupId> groupIds = colocateIndex.getAllGroupIds();
for (GroupId groupId : groupIds) {
ColocateGroupSchema groupSchema = colocateIndex.getGroupSchema(groupId);
if (groupSchema == null) {
LOG.info("Not found colocate group {}, maybe delete", groupId);
continue;
}
List<Long> tableIds = colocateIndex.getAllTableIds(groupId);
List<Set<Long>> backendBucketsSeq = colocateIndex.getBackendsPerBucketSeqSet(groupId);
if (backendBucketsSeq.isEmpty()) {
continue;
}
ColocateGroupSchema groupSchema = colocateIndex.getGroupSchema(groupId);
ReplicaAllocation replicaAlloc = groupSchema.getReplicaAlloc();
String unstableReason = null;
OUT:
@ -588,6 +597,7 @@ public class ColocateTableCheckerAndBalancer extends MasterDaemon {
for (GroupId groupId : groupIds) {
ColocateGroupSchema groupSchema = colocateIndex.getGroupSchema(groupId);
if (groupSchema == null) {
LOG.info("Not found colocate group {}, maybe delete", groupId);
continue;
}
ReplicaAllocation replicaAlloc = groupSchema.getReplicaAlloc();
@ -718,6 +728,10 @@ public class ColocateTableCheckerAndBalancer extends MasterDaemon {
GlobalColocateStatistic globalColocateStatistic, List<List<Long>> balancedBackendsPerBucketSeq,
boolean balanceBetweenGroups) {
ColocateGroupSchema groupSchema = colocateIndex.getGroupSchema(groupId);
if (groupSchema == null) {
LOG.info("Not found colocate group {}, maybe delete", groupId);
return false;
}
short replicaNum = groupSchema.getReplicaAlloc().getReplicaNumByTag(tag);
List<List<Long>> backendsPerBucketSeq = Lists.newArrayList(
colocateIndex.getBackendsPerBucketSeqByTag(groupId, tag));