From 8991c46dcc166ef04cc3f32a5c959267bdb13f6a Mon Sep 17 00:00:00 2001 From: deardeng Date: Mon, 10 Aug 2026 10:49:20 +0800 Subject: [PATCH] [improvement](fe) Presize global cloud tablet route sets pick from https://github.com/apache/doris/pull/66447 Presize cloud tablet route sets and reuse boxed route identifiers during rebuilds. (cherry picked from commit 634cbaaee977f66e4d0860db2376a0f879093e6b) --- .../doris/cloud/catalog/CloudReplica.java | 5 + .../cloud/catalog/CloudTabletRebalancer.java | 180 ++++++++++++------ .../doris/system/SystemInfoService.java | 4 + 3 files changed, 132 insertions(+), 57 deletions(-) diff --git a/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudReplica.java b/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudReplica.java index e6772c3c39d9f8..40391b65587d48 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudReplica.java +++ b/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudReplica.java @@ -294,6 +294,11 @@ public long getClusterPrimaryBackendId(String clusterId) { return -1L; } + Long getNonColocatedPrimaryBackendId(String clusterId) { + List backendIds = primaryClusterToBackends.get(clusterId); + return backendIds == null || backendIds.isEmpty() ? null : backendIds.get(0); + } + @Override public Map getClusterToBackendForProcDisplay( Map> computeGroupBackendCache) { diff --git a/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudTabletRebalancer.java b/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudTabletRebalancer.java index 69df53fcda94e1..63ee4688efc333 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudTabletRebalancer.java +++ b/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudTabletRebalancer.java @@ -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; @@ -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> DEFAULT_GLOBAL_TABLET_SET_FACTORY = + ignored -> ConcurrentHashMap.newKeySet(); private volatile ConcurrentHashMap> beToTabletsGlobal = new ConcurrentHashMap>(); @@ -308,7 +313,7 @@ public enum StatType { } @Getter - private class InfightTablet { + private static class InfightTablet { private final long tabletId; private final String clusterId; @@ -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(); } } @@ -433,45 +440,52 @@ private class TransferPairInfo { } public Set getSnapshotTabletsInPrimaryByBeId(Long beId) { - Set tabletIds = Sets.newHashSet(); Set tablets = beToTabletsGlobal.get(beId); - if (tablets != null) { - // Create a copy - for (Tablet tablet : new HashSet<>(tablets)) { - tabletIds.add(tablet.getId()); - } - } - Set colocateTablets = beToColocateTabletsGlobal.get(beId); - if (colocateTablets != null) { - // Create a copy - for (Tablet tablet : new HashSet<>(colocateTablets)) { - tabletIds.add(tablet.getId()); - } - } + Set tabletIds = newSnapshotTabletSet(tabletSetSize(tablets) + tabletSetSize(colocateTablets)); + addSnapshotTablets(tabletIds, tablets); + addSnapshotTablets(tabletIds, colocateTablets); return tabletIds; } public Set getSnapshotTabletsInSecondaryByBeId(Long beId) { - Set tabletIds = Sets.newHashSet(); Set tablets = beToTabletsGlobalInSecondary.get(beId); - if (tablets != null) { - // Create a copy - for (Tablet tablet : new HashSet<>(tablets)) { - tabletIds.add(tablet.getId()); - } - } + Set tabletIds = newSnapshotTabletSet(tabletSetSize(tablets)); + addSnapshotTablets(tabletIds, tablets); return tabletIds; } public Set getSnapshotTabletsInPrimaryAndSecondaryByBeId(Long beId) { - Set tabletIds = Sets.newHashSet(); - tabletIds.addAll(getSnapshotTabletsInPrimaryByBeId(beId)); - tabletIds.addAll(getSnapshotTabletsInSecondaryByBeId(beId)); + Set primaryTablets = beToTabletsGlobal.get(beId); + Set colocateTablets = beToColocateTabletsGlobal.get(beId); + Set secondaryTablets = beToTabletsGlobalInSecondary.get(beId); + int expectedSize = tabletSetSize(primaryTablets) + + tabletSetSize(colocateTablets) + tabletSetSize(secondaryTablets); + Set tabletIds = newSnapshotTabletSet(expectedSize); + addSnapshotTablets(tabletIds, primaryTablets); + addSnapshotTablets(tabletIds, colocateTablets); + addSnapshotTablets(tabletIds, secondaryTablets); return tabletIds; } + private static int tabletSetSize(Set tablets) { + return tablets == null ? 0 : tablets.size(); + } + + private static void addSnapshotTablets(Set snapshot, Set tablets) { + if (tablets != null) { + for (Tablet tablet : tablets) { + snapshot.add(tablet.getId()); + } + } + } + + @VisibleForTesting + protected Set newSnapshotTabletSet(int expectedSize) { + return Sets.newHashSetWithExpectedSize(expectedSize); + } + public int getTabletNumByBackendId(long beId) { Map> sourceMap = beToTabletsGlobal; ConcurrentHashMap> futureMap = futureBeToTabletsGlobal; @@ -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 beIds = new ArrayList(); - List tabletIds = new ArrayList(); + List tablets = index.getTablets(); boolean isColocated = Env.getCurrentColocateIndex().isColocateTable(table.getId()); - for (Tablet tablet : index.getTablets()) { + int routeCount = isColocated ? 0 : tablets.size(); + List beIds = newRouteInfoList(routeCount); + List 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()) { @@ -1033,14 +1051,28 @@ private boolean completeRouteInfo() { return true; } + @VisibleForTesting + protected List newRouteInfoList(int initialCapacity) { + return new ArrayList<>(initialCapacity); + } + public void fillBeToTablets(long be, long tableId, long partId, long indexId, Tablet tablet, ConcurrentHashMap> globalBeToTablets, ConcurrentHashMap>> beToTabletsInTable, ConcurrentHashMap>>> 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> globalTabletSetFactory, + ConcurrentHashMap> globalBeToTablets, + ConcurrentHashMap>> beToTabletsInTable, + ConcurrentHashMap>>> + partToTablets) { // global - globalBeToTablets.putIfAbsent(be, ConcurrentHashMap.newKeySet()); - globalBeToTablets.get(be).add(tablet); + globalBeToTablets.computeIfAbsent(be, globalTabletSetFactory).add(tablet); // table beToTabletsInTable.putIfAbsent(tableId, new ConcurrentHashMap>()); @@ -1057,6 +1089,23 @@ public void fillBeToTablets(long be, long tableId, long partId, long indexId, Ta beToTabletsOfIndex.get(be).add(tablet); } + private Function> newGlobalTabletSetFactory(Map> previousBeToTablets) { + Map> previousRoute = previousBeToTablets == null + ? Collections.emptyMap() : previousBeToTablets; + return be -> { + Set 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 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); @@ -1132,6 +1181,10 @@ private void flushExpiredWarmupBatches() { } public void statRouteInfo() { + // Reuse the previous route cardinalities as bounded allocation hints while rebuilding indexes. + Function> currentGlobalTabletSetFactory = newGlobalTabletSetFactory(beToTabletsGlobal); + Function> futureGlobalTabletSetFactory = + newGlobalTabletSetFactory(futureBeToTabletsGlobal); ConcurrentHashMap> tmpBeToTabletsGlobal = new ConcurrentHashMap>(); ConcurrentHashMap> tmpFutureBeToTabletsGlobal = new ConcurrentHashMap>(); ConcurrentHashMap> tmpBeToTabletsGlobalInSecondary @@ -1156,24 +1209,32 @@ public void statRouteInfo() { Map 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 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 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) { @@ -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 tablets = tmpBeToTabletsGlobalInSecondary + Set 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); } } diff --git a/fe/fe-core/src/main/java/org/apache/doris/system/SystemInfoService.java b/fe/fe-core/src/main/java/org/apache/doris/system/SystemInfoService.java index e36b9687fe9860..80fd62eeb873e1 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/system/SystemInfoService.java +++ b/fe/fe-core/src/main/java/org/apache/doris/system/SystemInfoService.java @@ -316,6 +316,10 @@ public Backend getBackend(long backendId) { return getAllClusterBackendsNoException().get(backendId); } + public Backend getBackendByIdWithBoxedId(Long backendId) { + return getAllClusterBackendsNoException().get(backendId); + } + public List getBackends(List backendIds) { List backends = Lists.newArrayList(); for (long backendId : backendIds) {