Skip to content

[improvement](fe) Release cloud tablet scheduling indexes after balance #66451

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
deardeng wants to merge 2 commits into
base: master
Choose a base branch
from
+127 −9
Open

[improvement](fe) Release cloud tablet scheduling indexes after balance #66451

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
a553e20
[improvement](fe) Release cloud tablet scheduling indexes after balance
deardeng Aug 4, 2026
df70ffb
[fix](fe) Release cloud tablet scheduling indexes on migration failures
deardeng Aug 5, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
28 changes: 20 additions & 8 deletions fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudTabletRebalancer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -514,16 +514,19 @@ protected void runAfterCatalogReady() {
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;
Expand Down Expand Up @@ -642,6 +645,15 @@ private void performBalancing() {
}
}

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;
Expand Down
108 changes: 107 additions & 1 deletion fe/fe-core/src/test/java/org/apache/doris/cloud/catalog/CloudTabletRebalancerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -83,7 +83,11 @@ private static class TestRebalancer extends CloudTabletRebalancer {
private final Set<Long> internalDbIds = new HashSet<>();

TestRebalancer() {
super(null);
this(null);
}

TestRebalancer(CloudSystemInfoService cloudSystemInfoService) {
super(cloudSystemInfoService);
}

void setInternalDbIds(Set<Long> ids) {
Expand Down Expand Up @@ -349,6 +353,107 @@ public void testWarmupRollbackReusesInflightBoxedTabletIdAfterRouteRebuild() thr
}
}

@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,
Expand Down Expand Up @@ -415,6 +520,7 @@ private static MockedStatic<Env> mockRouteEnvironment(Long dbId, Long tableId, L
Mockito.when(tablet.getReplicas()).thenReturn(Collections.singletonList(replica));
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);
Expand Down
Loading