Skip to content

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

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: master
Choose a base branch
from
Open

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

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 @@ -283,6 +283,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
81 changes: 68 additions & 13 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 << 20;
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 @@ -334,7 +339,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 + Objects.hashCode(clusterId);
}
}

Expand Down Expand Up @@ -964,10 +971,12 @@ 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
Expand All @@ -981,12 +990,14 @@ private boolean completeRouteInfo() {
}

// 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;
}
Expand Down Expand Up @@ -1053,6 +1064,11 @@ 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,
Expand All @@ -1067,8 +1083,18 @@ void fillBeToTablets(Long be, Long tableId, Long partId, Long indexId, Long tabl
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.computeIfAbsent(be, ignored -> ConcurrentHashMap.newKeySet()).add(tabletId);
globalBeToTablets.computeIfAbsent(be, globalTabletSetFactory).add(tabletId);

// table
ConcurrentHashMap<Long, Set<Long>> beToTabletsOfTable =
Expand All @@ -1083,6 +1109,23 @@ void fillBeToTablets(Long be, Long tableId, Long partId, Long indexId, Long tabl
beToTabletsOfIndex.computeIfAbsent(be, ignored -> ConcurrentHashMap.newKeySet()).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 @@ -1158,6 +1201,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 Down Expand Up @@ -1201,8 +1250,10 @@ public void statRouteInfo() {
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;
try {
Expand All @@ -1218,8 +1269,10 @@ public void statRouteInfo() {
continue;
}

Backend be = replica.getPrimaryBackend(cluster, false);
Long beId = be == null ? Long.valueOf(-1L) : Long.valueOf(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;
}
Expand All @@ -1232,14 +1285,16 @@ public void statRouteInfo() {
tablets.add(tabletId);
}

InfightTablet taskKey = new InfightTablet(tabletId, cluster);
InfightTask task = tabletToInfightTask.get(taskKey);
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, 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 @@ -335,6 +335,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
Loading