Skip to content

[improvement](fe) Presize global cloud tablet route sets (#66447) #67283

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
deardeng wants to merge 1 commit into
base: branch-4.1
Choose a base branch
from
+135 −48
Open

[improvement](fe) Presize global cloud tablet route sets (#66447) #67283

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudReplica.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -284,6 +284,10 @@ public long getClusterPrimaryBackendId(String clusterId) {
return primaryClusterToBackend.getOrDefault(clusterId, -1L);
}

Long getNonColocatedPrimaryBackendId(String clusterId) {
return primaryClusterToBackend.get(clusterId);
}

// For proc display only. In cloud mode a replica is hashed to a different BE in each
// compute group, so expose a clusterId -> backendId mapping; the proc display builds
// a separate bucket sequence per compute group from it so each group's sequence is
Expand Down
175 changes: 127 additions & 48 deletions fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudTabletRebalancer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -52,6 +52,7 @@
import org.apache.doris.thrift.TWarmUpCacheAsyncRequest;
import org.apache.doris.thrift.TWarmUpCacheAsyncResponse;

import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import com.google.common.base.Strings;
import com.google.common.collect.Sets;
Expand All @@ -77,10 +78,14 @@
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;
import java.util.stream.Collectors;

public class CloudTabletRebalancer extends MasterDaemon {
private static final Logger LOG = LogManager.getLogger(CloudTabletRebalancer.class);
private static final int MAX_GLOBAL_TABLET_SET_INITIAL_CAPACITY = 1 << 16;
private static final Function<Long, Set<Long>> DEFAULT_GLOBAL_TABLET_SET_FACTORY =
ignored -> ConcurrentHashMap.newKeySet();

private volatile ConcurrentHashMap<Long, Set<Long>> beToTabletsGlobal =
new ConcurrentHashMap<Long, Set<Long>>();
Expand Down Expand Up @@ -309,7 +314,7 @@ public enum StatType {
}

@Getter
private class InfightTablet {
private static class InfightTablet {
private final long tabletId;
private final String clusterId;

Expand All @@ -332,7 +337,9 @@ public boolean equals(Object o) {

@Override
public int hashCode() {
return Objects.hash(tabletId, clusterId);
int result = 1;
result = 31 * result + Long.hashCode(tabletId);
return 31 * result + clusterId.hashCode();
}
}

Expand Down Expand Up @@ -434,39 +441,50 @@ private class TransferPairInfo {
}

public Set<Long> getSnapshotTabletsInPrimaryByBeId(Long beId) {
Set<Long> tabletIds = Sets.newHashSet();
Set<Long> tablets = beToTabletsGlobal.get(beId);
if (tablets != null) {
// Create a copy
tabletIds.addAll(new HashSet<>(tablets));
}

Set<Long> colocateTablets = beToColocateTabletsGlobal.get(beId);
if (colocateTablets != null) {
// Create a copy
tabletIds.addAll(new HashSet<>(colocateTablets));
}
Set<Long> tabletIds = newSnapshotTabletSet(tabletSetSize(tablets) + tabletSetSize(colocateTablets));
addSnapshotTablets(tabletIds, tablets);
addSnapshotTablets(tabletIds, colocateTablets);

return tabletIds;
}

public Set<Long> getSnapshotTabletsInSecondaryByBeId(Long beId) {
Set<Long> tabletIds = Sets.newHashSet();
Set<Long> tablets = beToTabletsGlobalInSecondary.get(beId);
if (tablets != null) {
// Create a copy
tabletIds.addAll(new HashSet<>(tablets));
}
Set<Long> tabletIds = newSnapshotTabletSet(tabletSetSize(tablets));
addSnapshotTablets(tabletIds, tablets);
return tabletIds;
}

public Set<Long> getSnapshotTabletsInPrimaryAndSecondaryByBeId(Long beId) {
Set<Long> tabletIds = Sets.newHashSet();
tabletIds.addAll(getSnapshotTabletsInPrimaryByBeId(beId));
tabletIds.addAll(getSnapshotTabletsInSecondaryByBeId(beId));
Set<Long> primaryTablets = beToTabletsGlobal.get(beId);
Set<Long> colocateTablets = beToColocateTabletsGlobal.get(beId);
Set<Long> secondaryTablets = beToTabletsGlobalInSecondary.get(beId);
int expectedSize = tabletSetSize(primaryTablets)
+ tabletSetSize(colocateTablets) + tabletSetSize(secondaryTablets);
Set<Long> tabletIds = newSnapshotTabletSet(expectedSize);
addSnapshotTablets(tabletIds, primaryTablets);
addSnapshotTablets(tabletIds, colocateTablets);
addSnapshotTablets(tabletIds, secondaryTablets);
return tabletIds;
}

private static int tabletSetSize(Set<Long> tablets) {
return tablets == null ? 0 : tablets.size();
}

private static void addSnapshotTablets(Set<Long> snapshot, Set<Long> tablets) {
if (tablets != null) {
snapshot.addAll(tablets);
}
}

@VisibleForTesting
protected Set<Long> newSnapshotTabletSet(int expectedSize) {
return Sets.newHashSetWithExpectedSize(expectedSize);
}

public int getTabletNumByBackendId(long beId) {
Map<Long, Set<Long>> sourceMap = beToTabletsGlobal;
ConcurrentHashMap<Long, Set<Long>> futureMap = futureBeToTabletsGlobal;
Expand Down Expand Up @@ -939,34 +957,38 @@ private boolean completeRouteInfo() {
long needRehashDeadTime = System.currentTimeMillis() - Config.rehash_tablet_after_be_dead_seconds * 1000L;
loopCloudReplica((Database db, Table table, Partition partition, MaterializedIndex index, String cluster) -> {
boolean assigned = false;
List<Long> beIds = new ArrayList<Long>();
List<Long> tabletIds = new ArrayList<Long>();
List<Tablet> tablets = index.getTablets();
boolean isColocated = Env.getCurrentColocateIndex().isColocateTable(table.getId());
for (Tablet tablet : index.getTablets()) {
int routeCount = isColocated ? 0 : tablets.size();
List<Long> beIds = newRouteInfoList(routeCount);
List<Long> tabletIds = newRouteInfoList(routeCount);
for (Tablet tablet : tablets) {
for (Replica r : tablet.getReplicas()) {
CloudReplica replica = (CloudReplica) r;
// clean secondary map
replica.checkAndClearSecondaryClusterToBe(cluster, needRehashDeadTime);
InfightTablet taskKey = new InfightTablet(tablet.getId(), cluster);
// colocate table no need to update primary backends
if (isColocated) {
replica.clearClusterToBe(cluster);
tabletToInfightTask.remove(taskKey);
tabletToInfightTask.remove(new InfightTablet(tablet.getId(), cluster));
continue;
}

// primary backend is alive or dead not long
Backend be = replica.getPrimaryBackend(cluster, false);
Long primaryBeId = replica.getNonColocatedPrimaryBackendId(cluster);
Backend be = primaryBeId == null
? null : Env.getCurrentSystemInfo().getBackendByIdWithBoxedId(primaryBeId);
if (be != null && (be.isQueryAvailable()
|| (!be.isQueryDisabled()
// Compatible with older version upgrades, see https://github.com/apache/doris/pull/42986
&& (be.getLastUpdateMs() <= 0 || be.getLastUpdateMs() > needRehashDeadTime)))) {
beIds.add(be.getId());
beIds.add(primaryBeId);
tabletIds.add(tablet.getId());
continue;
}

// primary backend not available too long, change one
InfightTablet taskKey = new InfightTablet(tablet.getId(), cluster);
long beId = -1L;
be = replica.getSecondaryBackend(cluster);
if (be != null && be.isQueryAvailable()) {
Expand Down Expand Up @@ -1028,14 +1050,37 @@ private boolean completeRouteInfo() {
return true;
}

@VisibleForTesting
protected <T> List<T> newRouteInfoList(int initialCapacity) {
return new ArrayList<>(initialCapacity);
}

public void fillBeToTablets(long be, long tableId, long partId, long indexId, long tabletId,
ConcurrentHashMap<Long, Set<Long>> globalBeToTablets,
ConcurrentHashMap<Long, ConcurrentHashMap<Long, Set<Long>>> beToTabletsInTable,
ConcurrentHashMap<Long, ConcurrentHashMap<Long, ConcurrentHashMap<Long, Set<Long>>>>
partToTablets) {
fillBeToTablets(Long.valueOf(be), Long.valueOf(tableId), Long.valueOf(partId), Long.valueOf(indexId),
Long.valueOf(tabletId), globalBeToTablets, beToTabletsInTable, partToTablets);
}

void fillBeToTablets(Long be, Long tableId, Long partId, Long indexId, Long tabletId,
ConcurrentHashMap<Long, Set<Long>> globalBeToTablets,
ConcurrentHashMap<Long, ConcurrentHashMap<Long, Set<Long>>> beToTabletsInTable,
ConcurrentHashMap<Long, ConcurrentHashMap<Long, ConcurrentHashMap<Long, Set<Long>>>>
partToTablets) {
fillBeToTablets(be, tableId, partId, indexId, tabletId, DEFAULT_GLOBAL_TABLET_SET_FACTORY,
globalBeToTablets, beToTabletsInTable, partToTablets);
}

private void fillBeToTablets(Long be, Long tableId, Long partId, Long indexId, Long tabletId,
Function<Long, Set<Long>> globalTabletSetFactory,
ConcurrentHashMap<Long, Set<Long>> globalBeToTablets,
ConcurrentHashMap<Long, ConcurrentHashMap<Long, Set<Long>>> beToTabletsInTable,
ConcurrentHashMap<Long, ConcurrentHashMap<Long, ConcurrentHashMap<Long, Set<Long>>>>
partToTablets) {
// global
globalBeToTablets.putIfAbsent(be, ConcurrentHashMap.newKeySet());
globalBeToTablets.get(be).add(tabletId);
globalBeToTablets.computeIfAbsent(be, globalTabletSetFactory).add(tabletId);

// table
beToTabletsInTable.putIfAbsent(tableId, new ConcurrentHashMap<Long, Set<Long>>());
Expand All @@ -1052,6 +1097,23 @@ public void fillBeToTablets(long be, long tableId, long partId, long indexId, lo
beToTabletsOfIndex.get(be).add(tabletId);
}

private Function<Long, Set<Long>> newGlobalTabletSetFactory(Map<Long, Set<Long>> previousBeToTablets) {
Map<Long, Set<Long>> previousRoute = previousBeToTablets == null
? Collections.emptyMap() : previousBeToTablets;
return be -> {
Set<Long> previousTablets = previousRoute.get(be);
int initialCapacity = previousTablets == null ? 0
: Math.min(previousTablets.size(), MAX_GLOBAL_TABLET_SET_INITIAL_CAPACITY);
return newGlobalTabletSet(initialCapacity);
};
}

@VisibleForTesting
protected Set<Long> newGlobalTabletSet(int initialCapacity) {
return initialCapacity == 0
? ConcurrentHashMap.newKeySet() : ConcurrentHashMap.newKeySet(initialCapacity);
}

private void enqueueWarmupTask(WarmupTabletTask task) {
WarmupBatchKey key = new WarmupBatchKey(task.srcBe, task.destBe);
WarmupBatch batch = warmupBatches.computeIfAbsent(key, WarmupBatch::new);
Expand Down Expand Up @@ -1127,6 +1189,12 @@ private void flushExpiredWarmupBatches() {
}

public void statRouteInfo() {
// The previous generation remains live until the temporary global routes are complete, so reuse its
// per-backend cardinalities as allocation hints without extending its lifetime.
Function<Long, Set<Long>> currentGlobalTabletSetFactory =
newGlobalTabletSetFactory(beToTabletsGlobal);
Function<Long, Set<Long>> futureGlobalTabletSetFactory =
newGlobalTabletSetFactory(futureBeToTabletsGlobal);
ConcurrentHashMap<Long, Set<Long>> tmpBeToTabletsGlobal = new ConcurrentHashMap<Long, Set<Long>>();
ConcurrentHashMap<Long, Set<Long>> tmpFutureBeToTabletsGlobal = new ConcurrentHashMap<Long, Set<Long>>();
ConcurrentHashMap<Long, Set<Long>> tmpBeToTabletsGlobalInSecondary
Expand All @@ -1151,25 +1219,31 @@ public void statRouteInfo() {
Map<Long, Boolean> tmpDbInternal = new HashMap<>();

loopCloudReplica((Database db, Table table, Partition partition, MaterializedIndex index, String cluster) -> {
boolean isColocated = Env.getCurrentColocateIndex().isColocateTable(table.getId());
tmpTableToDb.put(table.getId(), db.getId());
tmpPartitionToDb.put(partition.getId(), db.getId());
tmpDbInternal.computeIfAbsent(db.getId(), k -> {
Long dbId = db.getId();
Long tableId = table.getId();
Long partitionId = partition.getId();
Long indexId = index.getId();
boolean isColocated = Env.getCurrentColocateIndex().isColocateTable(tableId);
tmpTableToDb.put(tableId, dbId);
tmpPartitionToDb.put(partitionId, dbId);
tmpDbInternal.computeIfAbsent(dbId, k -> {
String name = db.getFullName();
return name != null && INTERNAL_DB_NAMES.contains(name);
});
for (Tablet tablet : index.getTablets()) {
long tabletId = tablet.getId();
Long tabletId = tablet.getId();
// active tablet scoring (used for scheduling order)
if (activeTabletIds != null && !activeTabletIds.isEmpty() && activeTabletIds.contains(tabletId)) {
tmpTableActive.merge(table.getId(), 1L, Long::sum);
tmpPartitionActive.merge(partition.getId(), 1L, Long::sum);
tmpDbActive.merge(db.getId(), 1L, Long::sum);
tmpTableActive.merge(tableId, 1L, Long::sum);
tmpPartitionActive.merge(partitionId, 1L, Long::sum);
tmpDbActive.merge(dbId, 1L, Long::sum);
}
for (Replica r : tablet.getReplicas()) {
CloudReplica replica = (CloudReplica) r;
List<Replica> replicas = tablet.getReplicas();
int replicaCount = replicas.size();
for (int replicaIndex = 0; replicaIndex < replicaCount; replicaIndex++) {
CloudReplica replica = (CloudReplica) replicas.get(replicaIndex);
if (isColocated) {
long beId = -1L;
Long beId = -1L;
try {
beId = replica.getColocatedBeId(cluster);
} catch (ComputeGroupException e) {
Expand All @@ -1183,27 +1257,32 @@ public void statRouteInfo() {
continue;
}

Backend be = replica.getPrimaryBackend(cluster, false);
long beId = be == null ? -1L : be.getId();
Long primaryBeId = replica.getNonColocatedPrimaryBackendId(cluster);
Backend be = primaryBeId == null
? null : Env.getCurrentSystemInfo().getBackendByIdWithBoxedId(primaryBeId);
Long beId = be == null ? Long.valueOf(-1L) : primaryBeId;
if (!allBes.contains(beId)) {
continue;
}

Backend secondaryBe = replica.getSecondaryBackend(cluster);
long secondaryBeId = secondaryBe == null ? -1L : secondaryBe.getId();
Long secondaryBeId = secondaryBe == null ? Long.valueOf(-1L) : Long.valueOf(secondaryBe.getId());
if (allBes.contains(secondaryBeId)) {
Set<Long> tablets = tmpBeToTabletsGlobalInSecondary
.computeIfAbsent(secondaryBeId, k -> new HashSet<>());
tablets.add(tabletId);
}

InfightTablet taskKey = new InfightTablet(tabletId, cluster);
InfightTask task = tabletToInfightTask.get(taskKey);
long futureBeId = task == null ? beId : task.destBe;
fillBeToTablets(beId, table.getId(), partition.getId(), index.getId(), tabletId,
InfightTask task = tabletToInfightTask.isEmpty() ? null
: tabletToInfightTask.get(new InfightTablet(tabletId, cluster));
Long futureBeId = task == null ? beId : Long.valueOf(task.destBe);
Long routeTabletId = task == null ? tabletId : task.pickedTabletId;
fillBeToTablets(beId, tableId, partitionId, indexId, routeTabletId,
currentGlobalTabletSetFactory,
tmpBeToTabletsGlobal, beToTabletsInTable, this.partitionToTablets);

fillBeToTablets(futureBeId, table.getId(), partition.getId(), index.getId(), tabletId,
fillBeToTablets(futureBeId, tableId, partitionId, indexId, routeTabletId,
futureGlobalTabletSetFactory,
tmpFutureBeToTabletsGlobal, futureBeToTabletsInTable, futurePartitionToTablets);
}
}
Expand Down
4 changes: 4 additions & 0 deletions fe/fe-core/src/main/java/org/apache/doris/system/SystemInfoService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -336,6 +336,10 @@ public Backend getBackend(long backendId) {
return getAllClusterBackendsNoException().get(backendId);
}

public Backend getBackendByIdWithBoxedId(Long backendId) {
return getAllClusterBackendsNoException().get(backendId);
}

public List<Backend> getBackends(List<Long> backendIds) {
List<Backend> backends = Lists.newArrayList();
for (long backendId : backendIds) {
Expand Down
Loading