This is an automated email from the ASF dual-hosted git repository.
gavinchou pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new b0d6429032d [improvement](fe) Release cloud tablet scheduling indexes
after balance (#66451)
b0d6429032d is described below
commit b0d6429032dacd9b6fa23e5643ea591c14bae87c
Author: deardeng <[email protected]>
AuthorDate: Mon Aug 10 17:26:14 2026 +0800
[improvement](fe) Release cloud tablet scheduling indexes after balance
(#66451)
Related PR: #66378, #66389
Problem Summary: Cloud tablet route rebuilding retains the current and
future table-level and partition-level scheduling indexes for the entire
sleep interval after each balancing round, although later status checks
and external readers only need the global indexes. At 4 million tablets
across 4 clusters, a single-threaded JDK 17 path-level model estimates
that releasing these four nested graphs reduces approximate post-full-GC
retained route-index heap from 4.66 GiB to 1.35 GiB, a 3.31 GiB or
70.96% reduction. Cumulative route-index construction allocation remains
7.36 GiB, so this change reduces between-round retention rather than
allocation volume or in-round peak memory. Replace the four top-level
maps in a finally block after balancing so all exit paths release the
old graphs without an O(N) clear traversal, while preserving current and
future global routes. The next route-statistics pass rebuilds the
scheduling indexes before they are used again. These numbers are model
estimates, not production RSS measurements.
---
.../doris/cloud/catalog/CloudTabletRebalancer.java | 28 ++++--
.../cloud/catalog/CloudTabletRebalancerTest.java | 108 ++++++++++++++++++++-
2 files changed, 127 insertions(+), 9 deletions(-)
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 e65626e0f88..d4447e987fc 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
@@ -544,16 +544,19 @@ public class CloudTabletRebalancer extends MasterDaemon {
return;
}
- statRouteInfo();
- boolean migrated = migrateTabletsForSmoothUpgrade();
- if (migrated) {
+ try {
statRouteInfo();
- }
-
- indexBalanced = true;
- tableBalanced = true;
+ boolean migrated = migrateTabletsForSmoothUpgrade();
+ if (migrated) {
+ statRouteInfo();
+ }
- performBalancing();
+ indexBalanced = true;
+ tableBalanced = true;
+ performBalancing();
+ } finally {
+ releaseSchedulingIndexes();
+ }
checkDecommissionState(clusterToBes);
inited = true;
@@ -675,6 +678,15 @@ public class CloudTabletRebalancer extends MasterDaemon {
}
}
+ private void releaseSchedulingIndexes() {
+ // These indexes are no longer used after balancing. Replace their
top-level maps instead of clearing
+ // every entry so the complete tablet membership graphs can become
collectible without an O(N) traversal.
+ partitionToTablets = new ConcurrentHashMap<>();
+ futurePartitionToTablets = new ConcurrentHashMap<>();
+ beToTabletsInTable = new ConcurrentHashMap<>();
+ futureBeToTabletsInTable = new ConcurrentHashMap<>();
+ }
+
private boolean shouldForceInactivePhase(boolean activeBalanced) {
if (activeBalanced) {
consecutiveActiveUnbalancedRounds = 0;
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/cloud/catalog/CloudTabletRebalancerTest.java
b/fe/fe-core/src/test/java/org/apache/doris/cloud/catalog/CloudTabletRebalancerTest.java
index f183a28cfcc..be064ee442a 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/cloud/catalog/CloudTabletRebalancerTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/cloud/catalog/CloudTabletRebalancerTest.java
@@ -88,7 +88,11 @@ public class CloudTabletRebalancerTest {
private final Set<Long> internalDbIds = new HashSet<>();
TestRebalancer() {
- super(null);
+ this(null);
+ }
+
+ TestRebalancer(CloudSystemInfoService cloudSystemInfoService) {
+ super(cloudSystemInfoService);
}
void setInternalDbIds(Set<Long> ids) {
@@ -705,6 +709,107 @@ public class CloudTabletRebalancerTest {
Assertions.assertEquals(List.of(0, 0),
rebalancer.globalTabletSetInitialCapacities);
}
+ @Test
+ public void
testReleaseSchedulingIndexesKeepsGlobalRoutesAndAllowsNextRebuild() throws
Exception {
+ TestRebalancer rebalancer = new TestRebalancer();
+ Long srcBe = 10_001L;
+ Long dbId = 15_001L;
+ Long tableId = 20_001L;
+ Long partitionId = 30_001L;
+ Long indexId = 40_001L;
+ Long tabletId = 50_001L;
+ String clusterId = "cluster-a";
+ RouteMaps current = new RouteMaps();
+ RouteMaps future = new RouteMaps();
+ initializeRouteMaps(rebalancer, current, future, srcBe, tableId,
partitionId, indexId, tabletId);
+
+ invokePrivate(rebalancer, "releaseSchedulingIndexes", new Class<?>[]
{}, new Object[] {});
+
+ ConcurrentHashMap<Long, Set<Long>> currentGlobal =
getField(rebalancer, "beToTabletsGlobal");
+ ConcurrentHashMap<Long, Set<Long>> futureGlobal = getField(rebalancer,
"futureBeToTabletsGlobal");
+ ConcurrentHashMap<Long, ConcurrentHashMap<Long, Set<Long>>>
releasedCurrentByTable =
+ getField(rebalancer, "beToTabletsInTable");
+ ConcurrentHashMap<Long, ConcurrentHashMap<Long, Set<Long>>>
releasedFutureByTable =
+ getField(rebalancer, "futureBeToTabletsInTable");
+ ConcurrentHashMap<Long, ConcurrentHashMap<Long,
ConcurrentHashMap<Long, Set<Long>>>>
+ releasedCurrentByPartition = getField(rebalancer,
"partitionToTablets");
+ ConcurrentHashMap<Long, ConcurrentHashMap<Long,
ConcurrentHashMap<Long, Set<Long>>>>
+ releasedFutureByPartition = getField(rebalancer,
"futurePartitionToTablets");
+ Assertions.assertSame(current.global, currentGlobal);
+ Assertions.assertSame(future.global, futureGlobal);
+ Assertions.assertNotSame(current.byTable, releasedCurrentByTable);
+ Assertions.assertNotSame(future.byTable, releasedFutureByTable);
+ Assertions.assertNotSame(current.byPartition,
releasedCurrentByPartition);
+ Assertions.assertNotSame(future.byPartition,
releasedFutureByPartition);
+ Assertions.assertTrue(releasedCurrentByTable.isEmpty());
+ Assertions.assertTrue(releasedFutureByTable.isEmpty());
+ Assertions.assertTrue(releasedCurrentByPartition.isEmpty());
+ Assertions.assertTrue(releasedFutureByPartition.isEmpty());
+
+ setField(rebalancer, "clusterToBes",
Collections.singletonMap(clusterId, List.of(srcBe)));
+ setField(rebalancer, "allBes", Set.of(srcBe));
+ try (MockedStatic<Env> ignored = mockRouteEnvironment(
+ dbId, tableId, partitionId, indexId, tabletId, clusterId,
srcBe)) {
+ rebalancer.statRouteInfo();
+ }
+
+ ConcurrentHashMap<Long, ConcurrentHashMap<Long, Set<Long>>>
rebuiltCurrentByTable =
+ getField(rebalancer, "beToTabletsInTable");
+ ConcurrentHashMap<Long, ConcurrentHashMap<Long, Set<Long>>>
rebuiltFutureByTable =
+ getField(rebalancer, "futureBeToTabletsInTable");
+ ConcurrentHashMap<Long, ConcurrentHashMap<Long,
ConcurrentHashMap<Long, Set<Long>>>>
+ rebuiltCurrentByPartition = getField(rebalancer,
"partitionToTablets");
+ ConcurrentHashMap<Long, ConcurrentHashMap<Long,
ConcurrentHashMap<Long, Set<Long>>>>
+ rebuiltFutureByPartition = getField(rebalancer,
"futurePartitionToTablets");
+ Assertions.assertEquals(Set.of(tabletId),
rebuiltCurrentByTable.get(tableId).get(srcBe));
+ Assertions.assertEquals(Set.of(tabletId),
rebuiltFutureByTable.get(tableId).get(srcBe));
+ Assertions.assertEquals(Set.of(tabletId),
+
rebuiltCurrentByPartition.get(partitionId).get(indexId).get(srcBe));
+ Assertions.assertEquals(Set.of(tabletId),
+
rebuiltFutureByPartition.get(partitionId).get(indexId).get(srcBe));
+ }
+
+ @Test
+ public void
testRunAfterCatalogReadyReleasesSchedulingIndexesWhenMigrationFails() throws
Exception {
+ Long srcBe = 10_001L;
+ Long destBe = 10_002L;
+ Long dbId = 15_001L;
+ Long tableId = 20_001L;
+ Long partitionId = 30_001L;
+ Long indexId = 40_001L;
+ Long tabletId = 50_001L;
+ String clusterId = "cluster-a";
+ CloudSystemInfoService systemInfoService =
Mockito.mock(CloudSystemInfoService.class);
+ Backend srcBackend = Mockito.mock(Backend.class);
+
Mockito.when(systemInfoService.getAllBackendIds()).thenReturn(List.of(srcBe));
+
Mockito.when(systemInfoService.getBackend(srcBe)).thenReturn(srcBackend);
+ Mockito.when(srcBackend.getCloudClusterId()).thenReturn(clusterId);
+ TestRebalancer rebalancer = new TestRebalancer(systemInfoService);
+ rebalancer.addTabletMigrationTask(srcBe, destBe);
+
+ boolean oldEnableCloudMultiReplica = Config.enable_cloud_multi_replica;
+ Config.enable_cloud_multi_replica = false;
+ try (MockedStatic<Env> ignored = mockRouteEnvironment(
+ dbId, tableId, partitionId, indexId, tabletId, clusterId,
srcBe)) {
+ TabletInvertedIndex invertedIndex =
Env.getCurrentEnv().getTabletInvertedIndex();
+ Mockito.when(invertedIndex.getTabletMeta(tabletId))
+ .thenThrow(new RuntimeException("injected migration
failure"));
+
+ RuntimeException exception = Assertions.assertThrows(
+ RuntimeException.class, rebalancer::runAfterCatalogReady);
+
+ Assertions.assertEquals("injected migration failure",
exception.getMessage());
+ ConcurrentHashMap<Long, Set<Long>> currentGlobal =
getField(rebalancer, "beToTabletsGlobal");
+ Assertions.assertEquals(Set.of(tabletId),
currentGlobal.get(srcBe));
+ Assertions.assertTrue(((Map<?, ?>) getField(rebalancer,
"beToTabletsInTable")).isEmpty());
+ Assertions.assertTrue(((Map<?, ?>) getField(rebalancer,
"futureBeToTabletsInTable")).isEmpty());
+ Assertions.assertTrue(((Map<?, ?>) getField(rebalancer,
"partitionToTablets")).isEmpty());
+ Assertions.assertTrue(((Map<?, ?>) getField(rebalancer,
"futurePartitionToTablets")).isEmpty());
+ } finally {
+ Config.enable_cloud_multi_replica = oldEnableCloudMultiReplica;
+ }
+ }
+
private static void initializeRouteMaps(TestRebalancer rebalancer,
RouteMaps current, RouteMaps future,
Long srcBe, Long tableId, Long partitionId, Long indexId, Long
tabletId) throws Exception {
rebalancer.fillBeToTablets(srcBe, tableId, partitionId, indexId,
tabletId,
@@ -805,6 +910,7 @@ public class CloudTabletRebalancerTest {
Mockito.when(systemInfoService.getBackendByIdWithBoxedId(srcBe)).thenReturn(primaryBackend);
Mockito.when(replica.getPrimaryBackend(clusterId,
false)).thenReturn(primaryBackend);
Mockito.when(primaryBackend.getId()).thenReturn(srcBe);
+ Mockito.when(primaryBackend.isQueryAvailable()).thenReturn(true);
MockedStatic<Env> mockedEnv = Mockito.mockStatic(Env.class);
mockedEnv.when(Env::getCurrentEnv).thenReturn(env);
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
