Skip to content

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

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.0
Choose a base branch
from
Open

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

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
5 changes: 5 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 @@ -294,6 +294,11 @@ public long getClusterPrimaryBackendId(String clusterId) {
return -1L;
}

Long getNonColocatedPrimaryBackendId(String clusterId) {
List<Long> backendIds = primaryClusterToBackends.get(clusterId);
return backendIds == null || backendIds.isEmpty() ? null : backendIds.get(0);
}

@Override
public Map<String, Long> getClusterToBackendForProcDisplay(
Map<String, List<Backend>> computeGroupBackendCache) {
Expand Down
180 changes: 123 additions & 57 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 @@ -51,6 +51,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 @@ -76,10 +77,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<Tablet>> DEFAULT_GLOBAL_TABLET_SET_FACTORY =
ignored -> ConcurrentHashMap.newKeySet();

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

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

Expand All @@ -331,7 +336,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 @@ -433,45 +440,52 @@ private class TransferPairInfo {
}

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

Set<Tablet> colocateTablets = beToColocateTabletsGlobal.get(beId);
if (colocateTablets != null) {
// Create a copy
for (Tablet tablet : new HashSet<>(colocateTablets)) {
tabletIds.add(tablet.getId());
}
}
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<Tablet> tablets = beToTabletsGlobalInSecondary.get(beId);
if (tablets != null) {
// Create a copy
for (Tablet tablet : new HashSet<>(tablets)) {
tabletIds.add(tablet.getId());
}
}
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<Tablet> primaryTablets = beToTabletsGlobal.get(beId);
Set<Tablet> colocateTablets = beToColocateTabletsGlobal.get(beId);
Set<Tablet> 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<Tablet> tablets) {
return tablets == null ? 0 : tablets.size();
}

private static void addSnapshotTablets(Set<Long> snapshot, Set<Tablet> tablets) {
if (tablets != null) {
for (Tablet tablet : tablets) {
snapshot.add(tablet.getId());
}
}
}

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

public int getTabletNumByBackendId(long beId) {
Map<Long, Set<Tablet>> sourceMap = beToTabletsGlobal;
ConcurrentHashMap<Long, Set<Tablet>> futureMap = futureBeToTabletsGlobal;
Expand Down Expand Up @@ -944,34 +958,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 @@ -1033,14 +1051,28 @@ 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, Tablet tablet,
ConcurrentHashMap<Long, Set<Tablet>> globalBeToTablets,
ConcurrentHashMap<Long, ConcurrentHashMap<Long, Set<Tablet>>> beToTabletsInTable,
ConcurrentHashMap<Long, ConcurrentHashMap<Long, ConcurrentHashMap<Long, Set<Tablet>>>>
partToTablets) {
fillBeToTablets(Long.valueOf(be), Long.valueOf(tableId), Long.valueOf(partId), Long.valueOf(indexId), tablet,
DEFAULT_GLOBAL_TABLET_SET_FACTORY, globalBeToTablets, beToTabletsInTable, partToTablets);
}

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

// table
beToTabletsInTable.putIfAbsent(tableId, new ConcurrentHashMap<Long, Set<Tablet>>());
Expand All @@ -1057,6 +1089,23 @@ public void fillBeToTablets(long be, long tableId, long partId, long indexId, Ta
beToTabletsOfIndex.get(be).add(tablet);
}

private Function<Long, Set<Tablet>> newGlobalTabletSetFactory(Map<Long, Set<Tablet>> previousBeToTablets) {
Map<Long, Set<Tablet>> previousRoute = previousBeToTablets == null
? Collections.emptyMap() : previousBeToTablets;
return be -> {
Set<Tablet> 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<Tablet> 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 @@ -1132,6 +1181,10 @@ private void flushExpiredWarmupBatches() {
}

public void statRouteInfo() {
// Reuse the previous route cardinalities as bounded allocation hints while rebuilding indexes.
Function<Long, Set<Tablet>> currentGlobalTabletSetFactory = newGlobalTabletSetFactory(beToTabletsGlobal);
Function<Long, Set<Tablet>> futureGlobalTabletSetFactory =
newGlobalTabletSetFactory(futureBeToTabletsGlobal);
ConcurrentHashMap<Long, Set<Tablet>> tmpBeToTabletsGlobal = new ConcurrentHashMap<Long, Set<Tablet>>();
ConcurrentHashMap<Long, Set<Tablet>> tmpFutureBeToTabletsGlobal = new ConcurrentHashMap<Long, Set<Tablet>>();
ConcurrentHashMap<Long, Set<Tablet>> tmpBeToTabletsGlobalInSecondary
Expand All @@ -1156,24 +1209,32 @@ 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()) {
List<Tablet> tablets = index.getTablets();
for (Tablet tablet : tablets) {
Long tabletId = tablet.getId();
// active tablet scoring (used for scheduling order)
if (activeTabletIds != null && !activeTabletIds.isEmpty() && activeTabletIds.contains(tablet.getId())) {
tmpTableActive.merge(table.getId(), 1L, Long::sum);
tmpPartitionActive.merge(partition.getId(), 1L, Long::sum);
tmpDbActive.merge(db.getId(), 1L, Long::sum);
if (activeTabletIds != null && !activeTabletIds.isEmpty() && activeTabletIds.contains(tabletId)) {
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 @@ -1187,27 +1248,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<Tablet> tablets = tmpBeToTabletsGlobalInSecondary
Set<Tablet> secondaryTablets = tmpBeToTabletsGlobalInSecondary
.computeIfAbsent(secondaryBeId, k -> new HashSet<>());
tablets.add(tablet);
secondaryTablets.add(tablet);
}

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

fillBeToTablets(futureBeId, table.getId(), partition.getId(), index.getId(), tablet,
fillBeToTablets(futureBeId, tableId, partitionId, indexId, routeTablet,
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 @@ -316,6 +316,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