This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.1 by this push:
new fcd443eae34 [improvement](fe) Presize global cloud tablet route sets
(#66447) (#67283)
fcd443eae34 is described below
commit fcd443eae3426ea8bde954aa1f4bb942beca5c4c
Author: deardeng <[email protected]>
AuthorDate: Mon Aug 31 16:23:08 2026 +0800
[improvement](fe) Presize global cloud tablet route sets (#66447) (#67283)
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)
### 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:
https://github.com/apache/doris-website/pull/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 -->
---
.../apache/doris/cloud/catalog/CloudReplica.java | 4 +
.../doris/cloud/catalog/CloudTabletRebalancer.java | 175 +++++++++++++++------
.../org/apache/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 347416d972c..40d55aa2255 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 class CloudReplica extends Replica implements
GsonPostProcessable {
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 d1758eae96e..11989525c4f 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.TStatusCode;
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.Executors;
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 class CloudTabletRebalancer extends MasterDaemon {
}
@Getter
- private class InfightTablet {
+ private static class InfightTablet {
private final long tabletId;
private final String clusterId;
@@ -341,7 +346,9 @@ public class CloudTabletRebalancer extends MasterDaemon {
@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 @@ public class CloudTabletRebalancer extends MasterDaemon {
}
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 @@ public class CloudTabletRebalancer extends MasterDaemon {
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 @@ public class CloudTabletRebalancer extends MasterDaemon
{
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 class CloudTabletRebalancer extends MasterDaemon {
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 @@ public class CloudTabletRebalancer extends MasterDaemon {
}
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 class CloudTabletRebalancer extends MasterDaemon
{
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 class CloudTabletRebalancer extends MasterDaemon
{
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);
}
}
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 46a03183953..ae2f365cd16 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 class SystemInfoService {
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) {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
