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

pick from #66447

Presize cloud tablet route sets and reuse boxed route identifiers during
rebuilds.

(cherry picked from commit 634cbaa)

### What problem does this PR solve?

Issue Number: close #xxx

Related PR: #xxx

Problem Summary:

### Release note

None

### Check List (For Author)

- Test <!-- At least one of them must be included. -->
    - [ ] Regression test
    - [ ] Unit Test
    - [ ] Manual test (add detailed scripts or steps below)
    - [ ] No need to test or manual test. Explain why:
- [ ] This is a refactor/code format and no logic has been changed.
        - [ ] Previous test can cover this change.
        - [ ] No code files have been changed.
        - [ ] Other reason <!-- Add your reason?  -->

- Behavior changed:
    - [ ] No.
    - [ ] Yes. <!-- Explain the behavior change -->

- Does this need documentation?
    - [ ] No.
- [ ] Yes. <!-- Add document PR link here. eg:
apache/doris-website#1214 -->

### Check List (For Reviewer who merge this PR)

- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label <!-- Add branch pick label that this PR
should merge into -->

Author: deardeng

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

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<Long, Set<Long>> DEFAULT_GLOBAL_TABLET_SET_FACTORY =
+            ignored -> ConcurrentHashMap.newKeySet();
 
     private final CloudTabletRebalancerMetrics rebalancerMetrics;
     private long currentRoundTabletScanCount;
@@ -318,7 +323,7 @@ public enum StatType {
     }
 
     @Getter
-    private class InfightTablet {
+    private static class InfightTablet {
         private final long tabletId;
         private final String clusterId;
 
@@ -341,7 +346,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();
         }
     }
 
@@ -443,39 +450,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;
@@ -954,34 +972,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()) {
@@ -1043,14 +1065,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>>());
@@ -1067,6 +1112,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);
@@ -1142,6 +1204,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
@@ -1166,25 +1234,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) {
@@ -1198,27 +1272,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);
                 }
             }

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<Backend> getBackends(List<Long> backendIds) {
         List<Backend> backends = Lists.newArrayList();
         for (long backendId : backendIds) {