From b664ba94822ec9f1947b473c61c28a4d55555294 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 | 4 + .../cloud/catalog/CloudTabletRebalancer.java | 175 +++++++++++++----- .../doris/system/SystemInfoService.java | 4 + 3 files changed, 135 insertions(+), 48 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 347416d972c3b8..40d55aa2255b57 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 @@ -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 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 82d29f3d701f68..3aeefe96abd5e3 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 @@ -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; @@ -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> DEFAULT_GLOBAL_TABLET_SET_FACTORY = + ignored -> ConcurrentHashMap.newKeySet(); private volatile ConcurrentHashMap> beToTabletsGlobal = new ConcurrentHashMap>(); @@ -309,7 +314,7 @@ public enum StatType { } @Getter - private class InfightTablet { + private static class InfightTablet { private final long tabletId; private final String clusterId; @@ -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(); } } @@ -434,39 +441,50 @@ private class TransferPairInfo { } public Set getSnapshotTabletsInPrimaryByBeId(Long beId) { - Set tabletIds = Sets.newHashSet(); Set tablets = beToTabletsGlobal.get(beId); - if (tablets != null) { - // Create a copy - tabletIds.addAll(new HashSet<>(tablets)); - } - Set colocateTablets = beToColocateTabletsGlobal.get(beId); - if (colocateTablets != null) { - // Create a copy - tabletIds.addAll(new HashSet<>(colocateTablets)); - } + 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 - tabletIds.addAll(new HashSet<>(tablets)); - } + 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) { + snapshot.addAll(tablets); + } + } + + @VisibleForTesting + protected Set newSnapshotTabletSet(int expectedSize) { + return Sets.newHashSetWithExpectedSize(expectedSize); + } + public int getTabletNumByBackendId(long beId) { Map> sourceMap = beToTabletsGlobal; ConcurrentHashMap> futureMap = futureBeToTabletsGlobal; @@ -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 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()) { @@ -1028,14 +1050,37 @@ 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, long tabletId, ConcurrentHashMap> globalBeToTablets, ConcurrentHashMap>> beToTabletsInTable, ConcurrentHashMap>>> 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> globalBeToTablets, + ConcurrentHashMap>> beToTabletsInTable, + ConcurrentHashMap>>> + 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> globalTabletSetFactory, + ConcurrentHashMap> globalBeToTablets, + ConcurrentHashMap>> beToTabletsInTable, + ConcurrentHashMap>>> + partToTablets) { // global - globalBeToTablets.putIfAbsent(be, ConcurrentHashMap.newKeySet()); - globalBeToTablets.get(be).add(tabletId); + globalBeToTablets.computeIfAbsent(be, globalTabletSetFactory).add(tabletId); // table beToTabletsInTable.putIfAbsent(tableId, new ConcurrentHashMap>()); @@ -1052,6 +1097,23 @@ public void fillBeToTablets(long be, long tableId, long partId, long indexId, lo beToTabletsOfIndex.get(be).add(tabletId); } + 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); @@ -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> currentGlobalTabletSetFactory = + newGlobalTabletSetFactory(beToTabletsGlobal); + Function> futureGlobalTabletSetFactory = + newGlobalTabletSetFactory(futureBeToTabletsGlobal); ConcurrentHashMap> tmpBeToTabletsGlobal = new ConcurrentHashMap>(); ConcurrentHashMap> tmpFutureBeToTabletsGlobal = new ConcurrentHashMap>(); ConcurrentHashMap> tmpBeToTabletsGlobalInSecondary @@ -1151,25 +1219,31 @@ 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()) { - 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 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) { @@ -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 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); } } 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 46a0318395365f..ae2f365cd16181 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 @@ -336,6 +336,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) {