This is an automated email from the ASF dual-hosted git repository.
deardeng 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 53d6fa3de66 [improvement](fe) Reuse boxed IDs in cloud tablet indexes
(#66389)
53d6fa3de66 is described below
commit 53d6fa3de6698509d862eba9ac1d3ebc671aae85
Author: deardeng <[email protected]>
AuthorDate: Tue Aug 4 15:40:57 2026 +0800
[improvement](fe) Reuse boxed IDs in cloud tablet indexes (#66389)
Problem Summary: Cloud tablet route rebuilding repeatedly boxes
primitive backend, table, partition, index, and tablet IDs while
inserting the same logical IDs into current and future global, table,
and partition indexes.
This PR contains two related improvements:
1. Reuse boxed IDs across route indexes
Hoist boxing to the traversal and incremental update callers, then pass
the same immutable `Long` references through a boxed helper overload.
Preserve the public primitive method descriptor and the existing
`putIfAbsent` container implementation so the optimization
applies directly to master without depending on lazy container creation.
2. Preserve in-flight tablet IDs across route rebuilds
An asynchronous warmup task may remain in flight when the next rebalance
round rebuilds route information. Previously, the rebuild stored a newly
boxed tablet ID in the current and future indexes. If the warmup
subsequently failed, rollback restored the original boxed
ID held by the task only to the future indexes, while the current
indexes retained the newly boxed instance.
The fix reuses `InfightTask.pickedTabletId` when rebuilding both current
and future route indexes. This preserves boxed-ID identity across the
global, table-level, and partition-level indexes without changing
routing decisions, destination backends, or scheduling
semantics. It also prevents one additional `Long` wrapper from being
retained for each affected failed in-flight tablet until the next route
rebuild.
A single-threaded JDK 17 allocation model that keeps eager container
candidates in both variants estimates that 4 million tablets across 4
clusters reduce cumulative allocation from 25.47 GiB to 19.76 GiB
(22.42%) and approximate post-full-GC retained heap from 6.80 GiB
to 4.94 GiB (27.37%). These are path-level model estimates rather than
production RSS measurements.
The allocation estimates cover the boxed-ID reuse optimization. The
cross-generation warmup rollback fix provides an additional lifecycle
correctness guarantee and is not included in those estimates.
related PR #66378
---
.../doris/cloud/catalog/CloudTabletRebalancer.java | 68 +++--
.../cloud/catalog/CloudTabletRebalancerTest.java | 330 +++++++++++++++++++++
2 files changed, 371 insertions(+), 27 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 1a231170b9e..fde9c0d4850 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
@@ -339,7 +339,7 @@ public class CloudTabletRebalancer extends MasterDaemon {
}
private class InfightTask {
- public long pickedTabletId;
+ public Long pickedTabletId;
public long srcBe;
public long destBe;
public long startTimestamp;
@@ -375,12 +375,12 @@ public class CloudTabletRebalancer extends MasterDaemon {
}
private static class WarmupTabletTask {
- private final long pickedTabletId;
+ private final Long pickedTabletId;
private final long srcBe;
private final long destBe;
private final String clusterId;
- WarmupTabletTask(long pickedTabletId, long srcBe, long destBe, String
clusterId) {
+ WarmupTabletTask(Long pickedTabletId, long srcBe, long destBe, String
clusterId) {
this.pickedTabletId = pickedTabletId;
this.srcBe = srcBe;
this.destBe = destBe;
@@ -1058,6 +1058,15 @@ public class CloudTabletRebalancer extends MasterDaemon {
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) {
// global
globalBeToTablets.computeIfAbsent(be, ignored ->
ConcurrentHashMap.newKeySet()).add(tabletId);
@@ -1173,25 +1182,29 @@ 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;
if (isColocated) {
- long beId = -1L;
+ Long beId = -1L;
try {
beId = replica.getColocatedBeId(cluster);
} catch (ComputeGroupException e) {
@@ -1206,13 +1219,13 @@ public class CloudTabletRebalancer extends MasterDaemon
{
}
Backend be = replica.getPrimaryBackend(cluster, false);
- long beId = be == null ? -1L : be.getId();
+ Long beId = be == null ? Long.valueOf(-1L) :
Long.valueOf(be.getId());
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<>());
@@ -1221,11 +1234,12 @@ public class CloudTabletRebalancer extends MasterDaemon
{
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,
+ Long futureBeId = task == null ? beId :
Long.valueOf(task.destBe);
+ Long routeTabletId = task == null ? tabletId :
task.pickedTabletId;
+ fillBeToTablets(beId, tableId, partitionId, indexId,
routeTabletId,
tmpBeToTabletsGlobal, beToTabletsInTable,
this.partitionToTablets);
- fillBeToTablets(futureBeId, table.getId(),
partition.getId(), index.getId(), tabletId,
+ fillBeToTablets(futureBeId, tableId, partitionId, indexId,
routeTabletId,
tmpFutureBeToTabletsGlobal,
futureBeToTabletsInTable, futurePartitionToTablets);
}
}
@@ -1615,7 +1629,7 @@ public class CloudTabletRebalancer extends MasterDaemon {
}
}
- private void updateBeToTablets(long tabletId, long srcBe, long destBe,
+ private void updateBeToTablets(Long tabletId, Long srcBe, Long destBe,
ConcurrentHashMap<Long, Set<Long>>
globalBeToTablets,
ConcurrentHashMap<Long,
ConcurrentHashMap<Long, Set<Long>>> beToTabletsInTable,
ConcurrentHashMap<Long,
ConcurrentHashMap<Long, ConcurrentHashMap<Long,
@@ -1625,9 +1639,9 @@ public class CloudTabletRebalancer extends MasterDaemon {
LOG.warn("tablet {} meta not found in inverted index, skip
updateBeToTablets", tabletId);
return;
}
- long tableId = tabletMeta.getTableId();
- long partId = tabletMeta.getPartitionId();
- long indexId = tabletMeta.getIndexId();
+ Long tableId = tabletMeta.getTableId();
+ Long partId = tabletMeta.getPartitionId();
+ Long indexId = tabletMeta.getIndexId();
Set<Long> globalSrcTablets = globalBeToTablets.get(srcBe);
if (globalSrcTablets == null || !globalSrcTablets.remove(tabletId)) {
@@ -1655,8 +1669,8 @@ public class CloudTabletRebalancer extends MasterDaemon {
}
}
- fillBeToTablets(destBe, tableId, partId, indexId, tabletId,
globalBeToTablets, beToTabletsInTable,
- partToTablets);
+ fillBeToTablets(destBe, tableId, partId, indexId, tabletId,
globalBeToTablets,
+ beToTabletsInTable, partToTablets);
}
private void updateClusterToBeMap(long tabletId, long destBe, String
clusterId,
@@ -1912,8 +1926,8 @@ public class CloudTabletRebalancer extends MasterDaemon {
break; // no need balance
}
- long srcBe = pairInfo.srcBe;
- long destBe = pairInfo.destBe;
+ Long srcBe = pairInfo.srcBe;
+ Long destBe = pairInfo.destBe;
Long pickedTabletId = pickTabletPreferCold(srcBe,
beToTablets.get(srcBe),
this.activeTabletIds, pickedTabletIds);
@@ -2068,7 +2082,7 @@ public class CloudTabletRebalancer extends MasterDaemon {
return chosen;
}
- private boolean preheatAndUpdateTablet(long pickedTabletId, long srcBe,
long destBe, String clusterId,
+ private boolean preheatAndUpdateTablet(Long pickedTabletId, Long srcBe,
Long destBe, String clusterId,
BalanceType balanceType) {
Backend srcBackend = cloudSystemInfoService.getBackend(srcBe);
Backend destBackend = cloudSystemInfoService.getBackend(destBe);
@@ -2094,7 +2108,7 @@ public class CloudTabletRebalancer extends MasterDaemon {
return true;
}
- private boolean transferTablet(long pickedTabletId, long srcBe, long
destBe, String clusterId,
+ private boolean transferTablet(Long pickedTabletId, Long srcBe, Long
destBe, String clusterId,
BalanceType balanceType,
List<UpdateCloudReplicaInfo> infos) {
LOG.debug("transfer {} from {} to {}, cluster {}, type {}",
pickedTabletId, srcBe, destBe, clusterId, balanceType);
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 080ee4f5a34..520932b5fb6 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
@@ -17,9 +17,21 @@
package org.apache.doris.cloud.catalog;
+import org.apache.doris.catalog.ColocateTableIndex;
+import org.apache.doris.catalog.Database;
+import org.apache.doris.catalog.Env;
+import org.apache.doris.catalog.MaterializedIndex;
+import org.apache.doris.catalog.OlapTable;
+import org.apache.doris.catalog.Partition;
+import org.apache.doris.catalog.Tablet;
+import org.apache.doris.catalog.TabletInvertedIndex;
+import org.apache.doris.catalog.TabletMeta;
+import org.apache.doris.cloud.persist.UpdateCloudReplicaInfo;
import org.apache.doris.cloud.system.CloudSystemInfoService;
import org.apache.doris.common.Config;
+import org.apache.doris.datasource.InternalCatalog;
import org.apache.doris.metric.MetricRepo;
+import org.apache.doris.system.Backend;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Assertions;
@@ -48,12 +60,14 @@ public class CloudTabletRebalancerTest {
private boolean oldEnableActiveScheduling;
private long oldActiveTabletIdsRefreshIntervalSecond;
private int oldForceInactiveAfterRounds;
+ private int oldWarmupBatchSize;
@BeforeEach
public void setUp() {
oldEnableActiveScheduling =
Config.enable_cloud_active_tablet_priority_scheduling;
oldActiveTabletIdsRefreshIntervalSecond =
Config.cloud_active_tablet_ids_refresh_interval_second;
oldForceInactiveAfterRounds =
Config.cloud_active_unbalanced_force_inactive_after_rounds;
+ oldWarmupBatchSize = Config.cloud_warm_up_batch_size;
Config.enable_cloud_active_tablet_priority_scheduling = true;
}
@@ -62,6 +76,7 @@ public class CloudTabletRebalancerTest {
Config.enable_cloud_active_tablet_priority_scheduling =
oldEnableActiveScheduling;
Config.cloud_active_tablet_ids_refresh_interval_second =
oldActiveTabletIdsRefreshIntervalSecond;
Config.cloud_active_unbalanced_force_inactive_after_rounds =
oldForceInactiveAfterRounds;
+ Config.cloud_warm_up_batch_size = oldWarmupBatchSize;
}
private static class TestRebalancer extends CloudTabletRebalancer {
@@ -126,6 +141,321 @@ public class CloudTabletRebalancerTest {
return (T) m.invoke(obj, args);
}
+ @SuppressWarnings("unchecked")
+ private static <T> T invokePrivate(Object obj, String method, int
parameterCount, Object[] args)
+ throws Exception {
+ Method target = null;
+ for (Method candidate :
CloudTabletRebalancer.class.getDeclaredMethods()) {
+ if (candidate.getName().equals(method) &&
candidate.getParameterCount() == parameterCount) {
+ target = candidate;
+ break;
+ }
+ }
+ Assertions.assertNotNull(target, "Cannot find method " + method);
+ target.setAccessible(true);
+ return (T) target.invoke(obj, args);
+ }
+
+ private static class RouteMaps {
+ private final ConcurrentHashMap<Long, Set<Long>> global = new
ConcurrentHashMap<>();
+ private final ConcurrentHashMap<Long, ConcurrentHashMap<Long,
Set<Long>>> byTable =
+ new ConcurrentHashMap<>();
+ private final ConcurrentHashMap<Long, ConcurrentHashMap<Long,
ConcurrentHashMap<Long, Set<Long>>>>
+ byPartition = new ConcurrentHashMap<>();
+ }
+
+ @Test
+ public void testFillBeToTabletsReusesBoxedIdsAcrossIndexes() {
+ TestRebalancer rebalancer = new TestRebalancer();
+ Long beId = 10_001L;
+ Long tableId = 20_001L;
+ Long partitionId = 30_001L;
+ Long indexId = 40_001L;
+ Long tabletId = 50_001L;
+
+ ConcurrentHashMap<Long, Set<Long>> currentGlobal = new
ConcurrentHashMap<>();
+ ConcurrentHashMap<Long, ConcurrentHashMap<Long, Set<Long>>>
currentByTable =
+ new ConcurrentHashMap<>();
+ ConcurrentHashMap<Long, ConcurrentHashMap<Long,
ConcurrentHashMap<Long, Set<Long>>>> currentByPartition =
+ new ConcurrentHashMap<>();
+ ConcurrentHashMap<Long, Set<Long>> futureGlobal = new
ConcurrentHashMap<>();
+ ConcurrentHashMap<Long, ConcurrentHashMap<Long, Set<Long>>>
futureByTable =
+ new ConcurrentHashMap<>();
+ ConcurrentHashMap<Long, ConcurrentHashMap<Long,
ConcurrentHashMap<Long, Set<Long>>>> futureByPartition =
+ new ConcurrentHashMap<>();
+
+ rebalancer.fillBeToTablets(beId, tableId, partitionId, indexId,
tabletId,
+ currentGlobal, currentByTable, currentByPartition);
+ rebalancer.fillBeToTablets(beId, tableId, partitionId, indexId,
tabletId,
+ futureGlobal, futureByTable, futureByPartition);
+
+ assertSameStoredId(beId, currentGlobal);
+ assertSameStoredId(beId, currentByTable.get(tableId));
+ assertSameStoredId(beId,
currentByPartition.get(partitionId).get(indexId));
+ assertSameStoredId(beId, futureGlobal);
+ assertSameStoredId(beId, futureByTable.get(tableId));
+ assertSameStoredId(beId,
futureByPartition.get(partitionId).get(indexId));
+ assertSameStoredId(tableId, currentByTable);
+ assertSameStoredId(tableId, futureByTable);
+ assertSameStoredId(partitionId, currentByPartition);
+ assertSameStoredId(partitionId, futureByPartition);
+ assertSameStoredId(indexId, currentByPartition.get(partitionId));
+ assertSameStoredId(indexId, futureByPartition.get(partitionId));
+ assertSameStoredId(tabletId, currentGlobal.get(beId));
+ assertSameStoredId(tabletId, currentByTable.get(tableId).get(beId));
+ assertSameStoredId(tabletId,
currentByPartition.get(partitionId).get(indexId).get(beId));
+ assertSameStoredId(tabletId, futureGlobal.get(beId));
+ assertSameStoredId(tabletId, futureByTable.get(tableId).get(beId));
+ assertSameStoredId(tabletId,
futureByPartition.get(partitionId).get(indexId).get(beId));
+ }
+
+ @Test
+ public void
testTransferTabletReusesSelectedBoxedIdsAcrossCurrentAndFutureIndexes() throws
Exception {
+ TestRebalancer rebalancer = new TestRebalancer();
+ Long srcBe = 10_001L;
+ Long destBe = 10_002L;
+ Long tableId = 20_001L;
+ Long partitionId = 30_001L;
+ Long indexId = 40_001L;
+ Long tabletId = 50_001L;
+ RouteMaps current = new RouteMaps();
+ RouteMaps future = new RouteMaps();
+ initializeRouteMaps(rebalancer, current, future, srcBe, tableId,
partitionId, indexId, tabletId);
+
+ try (MockedStatic<Env> ignored = mockTabletMeta(tabletId, tableId,
partitionId, indexId)) {
+ boolean moved = invokePrivate(rebalancer, "transferTablet", 6,
+ new Object[] {tabletId, srcBe, destBe, "cluster-a",
+ CloudTabletRebalancer.BalanceType.GLOBAL, new
ArrayList<UpdateCloudReplicaInfo>()});
+
+ Assertions.assertTrue(moved);
+ assertSameRouteIds(destBe, tableId, partitionId, indexId,
tabletId, current);
+ assertSameRouteIds(destBe, tableId, partitionId, indexId,
tabletId, future);
+ }
+ }
+
+ @Test
+ public void testPreheatTabletReusesSelectedBoxedIdsInFutureIndexes()
throws Exception {
+ TestRebalancer rebalancer = new TestRebalancer();
+ Long srcBe = 10_001L;
+ Long destBe = 10_002L;
+ Long tableId = 20_001L;
+ Long partitionId = 30_001L;
+ Long indexId = 40_001L;
+ Long tabletId = 50_001L;
+ RouteMaps current = new RouteMaps();
+ RouteMaps future = new RouteMaps();
+ initializeRouteMaps(rebalancer, current, future, srcBe, tableId,
partitionId, indexId, tabletId);
+ setField(rebalancer, "cloudSystemInfoService",
mockBackendService(srcBe, destBe));
+ Config.cloud_warm_up_batch_size = 10;
+
+ try (MockedStatic<Env> ignored = mockTabletMeta(tabletId, tableId,
partitionId, indexId)) {
+ boolean moved = invokePrivate(rebalancer,
"preheatAndUpdateTablet", 5,
+ new Object[] {tabletId, srcBe, destBe, "cluster-a",
CloudTabletRebalancer.BalanceType.GLOBAL});
+
+ Assertions.assertTrue(moved);
+ assertSameRouteIds(destBe, tableId, partitionId, indexId,
tabletId, future);
+ }
+ }
+
+ @Test
+ public void testWarmupRollbackRestoresSelectedBoxedIdsInFutureIndexes()
throws Exception {
+ TestRebalancer rebalancer = new TestRebalancer();
+ Long srcBe = 10_001L;
+ Long destBe = 10_002L;
+ Long tableId = 20_001L;
+ Long partitionId = 30_001L;
+ Long indexId = 40_001L;
+ Long tabletId = 50_001L;
+ RouteMaps current = new RouteMaps();
+ RouteMaps future = new RouteMaps();
+ initializeRouteMaps(rebalancer, current, future, srcBe, tableId,
partitionId, indexId, tabletId);
+ setField(rebalancer, "cloudSystemInfoService",
mockBackendService(srcBe, destBe));
+ Config.cloud_warm_up_batch_size = 10;
+
+ try (MockedStatic<Env> ignored = mockTabletMeta(tabletId, tableId,
partitionId, indexId)) {
+ boolean moved = invokePrivate(rebalancer,
"preheatAndUpdateTablet", 5,
+ new Object[] {tabletId, srcBe, destBe, "cluster-a",
CloudTabletRebalancer.BalanceType.GLOBAL});
+ Assertions.assertTrue(moved);
+
+ Map<?, ?> warmupBatches = getField(rebalancer, "warmupBatches");
+ Object batch = warmupBatches.values().iterator().next();
+ Field tasksField = batch.getClass().getDeclaredField("tasks");
+ tasksField.setAccessible(true);
+ Object task = ((List<?>) tasksField.get(batch)).get(0);
+ invokePrivate(rebalancer, "revertWarmupState", new Class<?>[]
{task.getClass()}, new Object[] {task});
+
+ assertSameRouteIds(srcBe, tableId, partitionId, indexId, tabletId,
future);
+ }
+ }
+
+ @Test
+ public void
testWarmupRollbackReusesInflightBoxedTabletIdAfterRouteRebuild() throws
Exception {
+ TestRebalancer rebalancer = new TestRebalancer();
+ 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";
+ RouteMaps current = new RouteMaps();
+ RouteMaps future = new RouteMaps();
+ initializeRouteMaps(rebalancer, current, future, srcBe, tableId,
partitionId, indexId, tabletId);
+ setField(rebalancer, "cloudSystemInfoService",
mockBackendService(srcBe, destBe));
+ setField(rebalancer, "clusterToBes",
Collections.singletonMap(clusterId, List.of(srcBe, destBe)));
+ setField(rebalancer, "allBes", Set.of(srcBe, destBe));
+ Config.cloud_warm_up_batch_size = 10;
+
+ try (MockedStatic<Env> ignored = mockRouteEnvironment(
+ dbId, tableId, partitionId, indexId, tabletId, clusterId,
srcBe)) {
+ boolean moved = invokePrivate(rebalancer,
"preheatAndUpdateTablet", 5,
+ new Object[] {tabletId, srcBe, destBe, clusterId,
+ CloudTabletRebalancer.BalanceType.GLOBAL});
+ Assertions.assertTrue(moved);
+
+ Map<?, ?> warmupBatches = getField(rebalancer, "warmupBatches");
+ Object batch = warmupBatches.values().iterator().next();
+ Field tasksField = batch.getClass().getDeclaredField("tasks");
+ tasksField.setAccessible(true);
+ Object task = ((List<?>) tasksField.get(batch)).get(0);
+
+ rebalancer.statRouteInfo();
+ invokePrivate(rebalancer, "handleWarmupBatchFailure",
+ new Class<?>[] {List.class, Exception.class},
+ new Object[] {Collections.singletonList(task), null});
+ invokePrivate(rebalancer, "processFailedWarmupTasks", new
Class<?>[] {}, new Object[] {});
+
+ ConcurrentHashMap<Long, Set<Long>> rebuiltCurrentGlobal =
getField(rebalancer, "beToTabletsGlobal");
+ ConcurrentHashMap<Long, Set<Long>> rebuiltFutureGlobal = getField(
+ rebalancer, "futureBeToTabletsGlobal");
+ 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");
+ Long currentTabletId = getStoredId(tabletId,
rebuiltCurrentGlobal.get(srcBe));
+ Long futureTabletId = getStoredId(tabletId,
rebuiltFutureGlobal.get(srcBe));
+ Assertions.assertSame(currentTabletId, futureTabletId);
+ assertSameStoredId(currentTabletId,
rebuiltCurrentByTable.get(tableId).get(srcBe));
+ assertSameStoredId(currentTabletId,
rebuiltFutureByTable.get(tableId).get(srcBe));
+ assertSameStoredId(currentTabletId,
+
rebuiltCurrentByPartition.get(partitionId).get(indexId).get(srcBe));
+ assertSameStoredId(currentTabletId,
+
rebuiltFutureByPartition.get(partitionId).get(indexId).get(srcBe));
+ }
+ }
+
+ 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,
+ current.global, current.byTable, current.byPartition);
+ rebalancer.fillBeToTablets(srcBe, tableId, partitionId, indexId,
tabletId,
+ future.global, future.byTable, future.byPartition);
+ setField(rebalancer, "beToTabletsGlobal", current.global);
+ setField(rebalancer, "beToTabletsInTable", current.byTable);
+ setField(rebalancer, "partitionToTablets", current.byPartition);
+ setField(rebalancer, "futureBeToTabletsGlobal", future.global);
+ setField(rebalancer, "futureBeToTabletsInTable", future.byTable);
+ setField(rebalancer, "futurePartitionToTablets", future.byPartition);
+ }
+
+ private static MockedStatic<Env> mockTabletMeta(Long tabletId, Long
tableId, Long partitionId, Long indexId) {
+ Env env = Mockito.mock(Env.class);
+ TabletInvertedIndex invertedIndex =
Mockito.mock(TabletInvertedIndex.class);
+ TabletMeta tabletMeta = Mockito.mock(TabletMeta.class);
+ InternalCatalog catalog = Mockito.mock(InternalCatalog.class);
+ Mockito.when(env.getTabletInvertedIndex()).thenReturn(invertedIndex);
+
Mockito.when(invertedIndex.getTabletMeta(tabletId)).thenReturn(tabletMeta);
+ Mockito.when(tabletMeta.getTableId()).thenReturn(tableId);
+ Mockito.when(tabletMeta.getPartitionId()).thenReturn(partitionId);
+ Mockito.when(tabletMeta.getIndexId()).thenReturn(indexId);
+ MockedStatic<Env> mockedEnv = Mockito.mockStatic(Env.class);
+ mockedEnv.when(Env::getCurrentEnv).thenReturn(env);
+ mockedEnv.when(Env::getCurrentInternalCatalog).thenReturn(catalog);
+ return mockedEnv;
+ }
+
+ private static MockedStatic<Env> mockRouteEnvironment(Long dbId, Long
tableId, Long partitionId,
+ Long indexId, Long tabletId, String clusterId, Long srcBe) {
+ Env env = Mockito.mock(Env.class);
+ TabletInvertedIndex invertedIndex =
Mockito.mock(TabletInvertedIndex.class);
+ TabletMeta tabletMeta = Mockito.mock(TabletMeta.class);
+ InternalCatalog catalog = Mockito.mock(InternalCatalog.class);
+ ColocateTableIndex colocateTableIndex =
Mockito.mock(ColocateTableIndex.class);
+ Database database = Mockito.mock(Database.class);
+ OlapTable table = Mockito.mock(OlapTable.class);
+ Partition partition = Mockito.mock(Partition.class);
+ MaterializedIndex index = Mockito.mock(MaterializedIndex.class);
+ Tablet tablet = Mockito.mock(Tablet.class);
+ CloudReplica replica = Mockito.mock(CloudReplica.class);
+ Backend primaryBackend = Mockito.mock(Backend.class);
+
+ Mockito.when(env.getTabletInvertedIndex()).thenReturn(invertedIndex);
+
Mockito.when(invertedIndex.getTabletMeta(tabletId)).thenReturn(tabletMeta);
+ Mockito.when(tabletMeta.getTableId()).thenReturn(tableId);
+ Mockito.when(tabletMeta.getPartitionId()).thenReturn(partitionId);
+ Mockito.when(tabletMeta.getIndexId()).thenReturn(indexId);
+
Mockito.when(catalog.getDbIds()).thenReturn(Collections.singletonList(dbId));
+ Mockito.when(catalog.getDbNullable(dbId)).thenReturn(database);
+
Mockito.when(database.getTables()).thenReturn(Collections.singletonList(table));
+ Mockito.when(database.getId()).thenReturn(dbId);
+ Mockito.when(table.isManagedTable()).thenReturn(true);
+ Mockito.when(table.getId()).thenReturn(tableId);
+
Mockito.when(table.getAllPartitions()).thenReturn(Collections.singletonList(partition));
+ Mockito.when(partition.getId()).thenReturn(partitionId);
+
Mockito.when(partition.getMaterializedIndices(MaterializedIndex.IndexExtState.VISIBLE))
+ .thenReturn(Collections.singletonList(index));
+ Mockito.when(index.getId()).thenReturn(indexId);
+
Mockito.when(index.getTablets()).thenReturn(Collections.singletonList(tablet));
+ Mockito.when(tablet.getId()).thenReturn(tabletId);
+
Mockito.when(tablet.getReplicas()).thenReturn(Collections.singletonList(replica));
+ Mockito.when(replica.getPrimaryBackend(clusterId,
false)).thenReturn(primaryBackend);
+ Mockito.when(primaryBackend.getId()).thenReturn(srcBe);
+
+ MockedStatic<Env> mockedEnv = Mockito.mockStatic(Env.class);
+ mockedEnv.when(Env::getCurrentEnv).thenReturn(env);
+ mockedEnv.when(Env::getCurrentInternalCatalog).thenReturn(catalog);
+
mockedEnv.when(Env::getCurrentColocateIndex).thenReturn(colocateTableIndex);
+ return mockedEnv;
+ }
+
+ private static CloudSystemInfoService mockBackendService(Long srcBe, Long
destBe) {
+ CloudSystemInfoService systemInfoService =
Mockito.mock(CloudSystemInfoService.class);
+
Mockito.when(systemInfoService.getBackend(srcBe)).thenReturn(Mockito.mock(Backend.class));
+
Mockito.when(systemInfoService.getBackend(destBe)).thenReturn(Mockito.mock(Backend.class));
+ return systemInfoService;
+ }
+
+ private static void assertSameRouteIds(Long beId, Long tableId, Long
partitionId, Long indexId,
+ Long tabletId, RouteMaps routeMaps) {
+ assertSameStoredId(beId, routeMaps.global);
+ assertSameStoredId(beId, routeMaps.byTable.get(tableId));
+ assertSameStoredId(beId,
routeMaps.byPartition.get(partitionId).get(indexId));
+ assertSameStoredId(tableId, routeMaps.byTable);
+ assertSameStoredId(partitionId, routeMaps.byPartition);
+ assertSameStoredId(indexId, routeMaps.byPartition.get(partitionId));
+ assertSameStoredId(tabletId, routeMaps.global.get(beId));
+ assertSameStoredId(tabletId, routeMaps.byTable.get(tableId).get(beId));
+ assertSameStoredId(tabletId,
routeMaps.byPartition.get(partitionId).get(indexId).get(beId));
+ }
+
+ private static <V> void assertSameStoredId(Long expected, Map<Long, V>
map) {
+ Long stored =
map.keySet().stream().filter(expected::equals).findFirst().orElseThrow();
+ Assertions.assertSame(expected, stored);
+ }
+
+ private static void assertSameStoredId(Long expected, Set<Long> ids) {
+ Assertions.assertSame(expected, getStoredId(expected, ids));
+ }
+
+ private static Long getStoredId(Long expected, Set<Long> ids) {
+ return ids.stream().filter(expected::equals).findFirst().orElseThrow();
+ }
+
@Test
public void testFillBeToTabletsUsesComputedContainers() {
TestRebalancer rebalancer = new TestRebalancer();
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
