github-actions[bot] commented on code in PR #68995:
URL: https://github.com/apache/doris/pull/68995#discussion_r4255427264
##########
fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudTabletRebalancer.java:
##########
@@ -1775,48 +1796,37 @@ private void handleWarmupCompletion(InfightTask task,
String clusterId, boolean
}
}
- 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,
- Set<Long>>>> partToTablets) {
+ private void updateBeToTablets(Long tabletId, Long srcBe, Long destBe,
String clusterId,
+ RouteVariant variant,
ConcurrentHashMap<Long, Set<Long>> globalBeToTablets) {
TabletMeta tabletMeta =
Env.getCurrentEnv().getTabletInvertedIndex().getTabletMeta(tabletId);
if (tabletMeta == null) {
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();
-
Set<Long> globalSrcTablets = globalBeToTablets.get(srcBe);
if (globalSrcTablets == null || !globalSrcTablets.remove(tabletId)) {
LOG.debug("skip updateBeToTablets for tablet {}: srcBe {} not in
globalBeToTablets", tabletId, srcBe);
return;
}
- ConcurrentHashMap<Long, Set<Long>> tableBeMap =
beToTabletsInTable.get(tableId);
- if (tableBeMap == null) {
- return;
- }
- Set<Long> tableSrcTablets = tableBeMap.get(srcBe);
- if (tableSrcTablets != null) {
- tableSrcTablets.remove(tabletId);
+ fillGlobalBeToTablets(destBe, tabletId,
DEFAULT_GLOBAL_TABLET_SET_FACTORY, globalBeToTablets);
+ if (clusterId.equals(tableCountClusterId)) {
+ Map<Long, Integer> counts = (variant == RouteVariant.CURRENT ?
currentTableCounts : futureTableCounts)
+ .get(tableId);
+ if (counts != null) {
+ counts.merge(srcBe, -1, Integer::sum);
+ counts.merge(destBe, 1, Integer::sum);
+ }
}
-
- ConcurrentHashMap<Long, ConcurrentHashMap<Long, Set<Long>>> indexMap =
partToTablets.get(partId);
- if (indexMap != null) {
- ConcurrentHashMap<Long, Set<Long>> beMap = indexMap.get(indexId);
- if (beMap != null) {
- Set<Long> partSrcTablets = beMap.get(srcBe);
- if (partSrcTablets != null) {
- partSrcTablets.remove(tabletId);
- }
+ if (clusterId.equals(indexCountClusterId)) {
Review Comment:
[P2] Keep live count updates consistent with canceled warmups
These cache updates run only after removal from the round-global source set
succeeds, but the new local views/counts read newer inflight state. The
independent checker removes warmups to a vanished/dead destination without
rolling back `futureBeToTabletsGlobal`. For example, use one partition with a
base index and five distinct rollups on A/B/C; each index has FUTURE counts
`(2,2,2)`, with its two B tablets actually primary-A tablets warming to B.
Partition balancing skips these balanced views. If B disappears and the checker
cancels those tasks before TABLE builds its view, the fresh table has A=24/C=12
and the default budget allows two attempts. Selecting both canceled tablets of
one index schedules A→C twice: their IDs remain globally at B, so
`updateBeToTablets` returns at the source-removal check and the cached index
counts stay `(4,2)`. The second move therefore passes instead of conflicting at
`(3,3)`, leaving that index at `(2,4)` and needing corrective balancing. Serial
ize cancellation through daemon rollback or reconcile global membership with
rebuilt views/counts before scheduling; a successful local move must advance
its resident count cache even when global membership came from an older
snapshot.
##########
fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudTabletRebalancer.java:
##########
@@ -1983,79 +1991,105 @@ private boolean isConflict(long srcBe, long destBe,
long tabletId, BalanceType b
return true; // Cannot find metadata, treat as conflict
}
- if (balanceType == BalanceType.GLOBAL) {
- return checkGlobalBalanceConflict(srcBe, destBe, tabletMeta,
beToTabletsInParts, beToTabletsInTables);
- } else if (balanceType == BalanceType.TABLE) {
- return checkTableBalanceConflict(srcBe, destBe, tabletMeta,
beToTabletsInParts);
+ if (balanceType == BalanceType.PARTITION) {
+ return false;
}
-
- return false;
- }
-
- private boolean checkGlobalBalanceConflict(long srcBe, long destBe,
TabletMeta tabletMeta,
- ConcurrentHashMap<Long,
- ConcurrentHashMap<Long,
ConcurrentHashMap<Long, Set<Long>>>>
- beToTabletsInParts,
- ConcurrentHashMap<Long,
ConcurrentHashMap<Long, Set<Long>>>
- beToTabletsInTables) {
- long maxBeSize = getTabletSizeInParts(srcBe, tabletMeta,
beToTabletsInParts);
- long minBeSize = getTabletSizeInParts(destBe, tabletMeta,
beToTabletsInParts);
-
- if (minBeSize >= maxBeSize) {
- return true; // Conflict detected
+ Database db =
Env.getCurrentInternalCatalog().getDbNullable(tabletMeta.getDbId());
+ if (db == null) {
+ return true;
}
-
- maxBeSize = getTabletSizeInBes(srcBe, tabletMeta, beToTabletsInTables);
- minBeSize = getTabletSizeInBes(destBe, tabletMeta,
beToTabletsInTables);
-
- return minBeSize >= maxBeSize; // Conflict detected
- }
-
- private boolean checkTableBalanceConflict(long srcBe, long destBe,
TabletMeta tabletMeta,
- ConcurrentHashMap<Long,
- ConcurrentHashMap<Long,
ConcurrentHashMap<Long, Set<Long>>>>
- beToTabletsInParts) {
- long maxBeSize = getTabletSizeInParts(srcBe, tabletMeta,
beToTabletsInParts);
- long minBeSize = getTabletSizeInParts(destBe, tabletMeta,
beToTabletsInParts);
-
- return minBeSize >= maxBeSize; // Conflict detected
- }
-
- private long getTabletSizeInParts(long beId, TabletMeta tabletMeta,
- ConcurrentHashMap<Long,
- ConcurrentHashMap<Long,
ConcurrentHashMap<Long, Set<Long>>>>
- beToTabletsInParts) {
- ConcurrentHashMap<Long, ConcurrentHashMap<Long, Set<Long>>>
indexToTablets
- = beToTabletsInParts.get(tabletMeta.getPartitionId());
- if (indexToTablets == null) {
- return 0;
+ OlapTable table = (OlapTable)
db.getTableNullable(tabletMeta.getTableId());
+ if (table == null) {
+ return true;
}
- ConcurrentHashMap<Long, Set<Long>> beToTablets =
indexToTablets.get(tabletMeta.getIndexId());
- if (beToTablets == null) {
- return 0;
+ table.readLock();
+ try {
+ Partition partition =
table.getPartition(tabletMeta.getPartitionId());
+ MaterializedIndex index = null;
+ if (partition != null) {
+ for (MaterializedIndex visibleIndex
+ :
partition.getMaterializedIndices(IndexExtState.VISIBLE, true)) {
+ if (visibleIndex.getId() == tabletMeta.getIndexId()) {
+ index = visibleIndex;
+ break;
+ }
+ }
+ }
+ if (index == null) {
+ return true;
+ }
+ Preconditions.checkState(clusterId.equals(indexCountClusterId));
+ Map<Long, Map<Long, Integer>> indexCache = variant ==
RouteVariant.CURRENT
+ ? currentIndexCounts : futureIndexCounts;
+ Map<Long, Integer> indexCounts = indexCache.get(index.getId());
Review Comment:
[P1] Include partition identity in the index-count cache key
A base/rollup index ID is reused by every partition of its table
(`InternalCatalog.createPartitionWithIndices` constructs each base index with
`tbl.getBaseIndexId()`), while this cache survives the whole TABLE/GLOBAL pass.
Therefore a candidate in another partition reuses the first partition's counts,
and `updateBeToTablets` also updates that unrelated entry. This can worsen
partition balance under the default settings: on four BEs A/B/C/D, take five
partitions with counts `(2,2,1,1)` and one with `(2,1,2,1)`. All partition
spreads are 1, so partition balancing skips them. A first TABLE move A→D from a
`(2,2,1,1)` partition changes the cached counts to `(1,2,1,2)` and table totals
to `(11,11,7,7)`. With B selected as the tied source and C as destination, the
cache then permits moving the other partition's B tablet to C, although that
partition actually has B=1/C=2; it becomes `(2,0,3,1)` and requires a
corrective partition move next round. Key both lookup and transfer/rollback u
pdates by `(partitionId,indexId)`. The new convergence test uses a different
`indexId` for each partition, so it does not cover this production case.
##########
fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudTabletRebalancer.java:
##########
@@ -1437,140 +1410,189 @@ public void loopCloudReplica(Operator operator) {
private void balanceInPartition(List<Long> bes, String clusterId,
List<UpdateCloudReplicaInfo> infos,
ActiveSchedulePhase phase) {
- // balance all partition (prefer active partitions/tables, put
internal db at tail)
- Iterable<Map.Entry<Long, ConcurrentHashMap<Long,
ConcurrentHashMap<Long, Set<Long>>>>> partitions;
+ Iterable<Long> partitions;
if (Config.enable_cloud_active_tablet_priority_scheduling) {
- final Comparator<Map.Entry<Long, ConcurrentHashMap<Long,
ConcurrentHashMap<Long, Set<Long>>>>> cmp =
- partitionEntryComparator();
- // Phase-aware filtering and ordering.
- // - ACTIVE_ONLY: only non-internal partitions with activeCnt > 0
- // - INACTIVE_ONLY: all remaining partitions (non-internal
inactive first, internal last)
- // - ALL: active (TopN first if configured) -> inactive -> internal
- List<Map.Entry<Long, ConcurrentHashMap<Long,
ConcurrentHashMap<Long, Set<Long>>>>> nonInternalActive =
- new ArrayList<>();
- List<Map.Entry<Long, ConcurrentHashMap<Long,
ConcurrentHashMap<Long, Set<Long>>>>> nonInternalInactive =
- new ArrayList<>();
- List<Map.Entry<Long, ConcurrentHashMap<Long,
ConcurrentHashMap<Long, Set<Long>>>>> internalPartitions =
- new ArrayList<>();
-
- for (Map.Entry<Long, ConcurrentHashMap<Long,
ConcurrentHashMap<Long, Set<Long>>>> e
- : futurePartitionToTablets.entrySet()) {
- long partId = e.getKey();
- boolean internal =
isInternalDbId(partitionIdToDbId.get(partId));
- long activeCnt = partitionIdToActiveCount.getOrDefault(partId,
0L);
-
- if (internal) {
- // internal partitions are always handled at the end (not
in ACTIVE_ONLY).
- internalPartitions.add(e);
- continue;
- }
-
- if (activeCnt > 0) {
- nonInternalActive.add(e);
+ Comparator<Long> cmp = partitionEntryComparator();
+ List<Long> nonInternalActive = new ArrayList<>();
+ List<Long> nonInternalInactive = new ArrayList<>();
+ List<Long> internalPartitions = new ArrayList<>();
+ for (Long partitionId : partitionIdToTableId.keySet()) {
+ if (isInternalDbId(partitionIdToDbId.get(partitionId))) {
+ internalPartitions.add(partitionId);
+ } else if (partitionIdToActiveCount.getOrDefault(partitionId,
0L) > 0) {
+ nonInternalActive.add(partitionId);
} else {
- nonInternalInactive.add(e);
+ nonInternalInactive.add(partitionId);
}
}
-
nonInternalActive.sort(cmp);
nonInternalInactive.sort(cmp);
internalPartitions.sort(cmp);
-
- List<Map.Entry<Long, ConcurrentHashMap<Long,
ConcurrentHashMap<Long, Set<Long>>>>> ordered =
- new ArrayList<>(futurePartitionToTablets.size());
- if (phase == ActiveSchedulePhase.ACTIVE_ONLY) {
- // In ACTIVE_ONLY phase, schedule all active partitions
(already sorted by cmp, most active first)
- ordered.addAll(nonInternalActive);
- } else if (phase == ActiveSchedulePhase.INACTIVE_ONLY) {
- ordered.addAll(nonInternalInactive);
- ordered.addAll(internalPartitions);
- } else { // ALL
- // All active (already sorted by cmp, most active first), then
inactive, then internal
+ List<Long> ordered = new ArrayList<>(partitionIdToTableId.size());
+ if (phase != ActiveSchedulePhase.INACTIVE_ONLY) {
ordered.addAll(nonInternalActive);
+ }
+ if (phase != ActiveSchedulePhase.ACTIVE_ONLY) {
ordered.addAll(nonInternalInactive);
ordered.addAll(internalPartitions);
}
-
partitions = ordered;
} else {
- partitions = futurePartitionToTablets.entrySet();
- }
-
- for (Map.Entry<Long, ConcurrentHashMap<Long, ConcurrentHashMap<Long,
Set<Long>>>> partitionEntry
- : partitions) {
- Map<Long, ConcurrentHashMap<Long, Set<Long>>> indexToTablets =
partitionEntry.getValue();
- // balance all index of a partition
- List<Map.Entry<Long, ConcurrentHashMap<Long, Set<Long>>>> indexes =
- new ArrayList<>(indexToTablets.entrySet());
- // index-level ordering is not critical; keep stable by id
- indexes.sort(Comparator.comparingLong(Map.Entry::getKey));
- for (Map.Entry<Long, ConcurrentHashMap<Long, Set<Long>>> entry :
indexes) {
- // balance a index
- // Fast path: this index has no tablets in this cluster, skip
to avoid useless balanceImpl work.
- if (calculateTotalTablets(bes, entry.getValue()) == 0) {
- continue;
+ partitions = partitionIdToTableId.keySet();
+ }
+
+ for (Long partitionId : partitions) {
+ processFailedWarmupTasks();
+ Map<Long, Map<Long, Set<Long>>> indexViews =
buildPartitionViews(partitionId, clusterId);
+ for (Long indexId : indexViews.keySet()) {
+ if (processFailedWarmupTasks()) {
+ indexViews = buildPartitionViews(partitionId, clusterId);
+ }
+ Map<Long, Set<Long>> view = indexViews.get(indexId);
+ if (view != null && calculateTotalTablets(bes, view) != 0) {
+ balanceImpl(bes, clusterId, view, BalanceType.PARTITION,
infos);
}
- balanceImpl(bes, clusterId, entry.getValue(),
BalanceType.PARTITION, infos);
}
}
}
private void balanceInTable(List<Long> bes, String clusterId,
List<UpdateCloudReplicaInfo> infos,
ActiveSchedulePhase phase) {
- // balance all tables (prefer active tables/dbs, put internal db at
tail)
- Iterable<Map.Entry<Long, ConcurrentHashMap<Long, Set<Long>>>> tables;
+ Iterable<Long> tables;
+ Set<Long> tableIds = new HashSet<>(partitionIdToTableId.values());
if (Config.enable_cloud_active_tablet_priority_scheduling) {
- final Comparator<Map.Entry<Long, ConcurrentHashMap<Long,
Set<Long>>>> cmp = tableEntryComparator();
- List<Map.Entry<Long, ConcurrentHashMap<Long, Set<Long>>>>
nonInternalActive = new ArrayList<>();
- List<Map.Entry<Long, ConcurrentHashMap<Long, Set<Long>>>>
nonInternalInactive = new ArrayList<>();
- List<Map.Entry<Long, ConcurrentHashMap<Long, Set<Long>>>>
internalTables = new ArrayList<>();
-
- for (Map.Entry<Long, ConcurrentHashMap<Long, Set<Long>>> e :
futureBeToTabletsInTable.entrySet()) {
- long tableId = e.getKey();
- boolean internal = isInternalDbId(tableIdToDbId.get(tableId));
- long activeCnt = tableIdToActiveCount.getOrDefault(tableId,
0L);
- if (internal) {
- internalTables.add(e);
- continue;
- }
- if (activeCnt > 0) {
- nonInternalActive.add(e);
+ Comparator<Long> cmp = tableEntryComparator();
+ List<Long> nonInternalActive = new ArrayList<>();
+ List<Long> nonInternalInactive = new ArrayList<>();
+ List<Long> internalTables = new ArrayList<>();
+ for (Long tableId : tableIds) {
+ if (isInternalDbId(tableIdToDbId.get(tableId))) {
+ internalTables.add(tableId);
+ } else if (tableIdToActiveCount.getOrDefault(tableId, 0L) > 0)
{
+ nonInternalActive.add(tableId);
} else {
- nonInternalInactive.add(e);
+ nonInternalInactive.add(tableId);
}
}
-
nonInternalActive.sort(cmp);
nonInternalInactive.sort(cmp);
internalTables.sort(cmp);
-
- List<Map.Entry<Long, ConcurrentHashMap<Long, Set<Long>>>> ordered =
- new ArrayList<>(futureBeToTabletsInTable.size());
- if (phase == ActiveSchedulePhase.ACTIVE_ONLY) {
- ordered.addAll(nonInternalActive);
- } else if (phase == ActiveSchedulePhase.INACTIVE_ONLY) {
- ordered.addAll(nonInternalInactive);
- ordered.addAll(internalTables);
- } else { // ALL
+ List<Long> ordered = new ArrayList<>(tableIds.size());
+ if (phase != ActiveSchedulePhase.INACTIVE_ONLY) {
ordered.addAll(nonInternalActive);
+ }
+ if (phase != ActiveSchedulePhase.ACTIVE_ONLY) {
ordered.addAll(nonInternalInactive);
ordered.addAll(internalTables);
}
-
tables = ordered;
} else {
- tables = futureBeToTabletsInTable.entrySet();
+ tables = tableIds;
}
- for (Map.Entry<Long, ConcurrentHashMap<Long, Set<Long>>> entry :
tables) {
- // Fast path: this table has no tablets in this cluster, skip.
- if (calculateTotalTablets(bes, entry.getValue()) == 0) {
- continue;
+ for (Long tableId : tables) {
+ processFailedWarmupTasks();
+ Map<Long, Set<Long>> view = buildTableView(tableId, clusterId);
+ if (calculateTotalTablets(bes, view) != 0) {
+ balanceWithIndexCounts(bes, clusterId, view,
BalanceType.TABLE, infos);
}
- balanceImpl(bes, clusterId, entry.getValue(), BalanceType.TABLE,
infos);
}
}
+ private long currentRouteBe(CloudReplica replica, String clusterId) {
+ Long beId = replica.getNonColocatedPrimaryBackendId(clusterId);
+ if (beId == null ||
Env.getCurrentSystemInfo().getBackendByIdWithBoxedId(beId) == null
+ || !allBes.contains(beId)) {
+ return -1L;
+ }
+ return beId;
+ }
+
+ private long futureRouteBe(long tabletId, CloudReplica replica, String
clusterId) {
+ long currentBe = currentRouteBe(replica, clusterId);
+ if (currentBe == -1L || tabletToInfightTask.isEmpty()) {
+ return currentBe;
+ }
+ InfightTask task = tabletToInfightTask.get(new InfightTablet(tabletId,
clusterId));
+ return task == null ? currentBe : task.destBe;
+ }
+
+ private void addFutureTablet(Map<Long, Set<Long>> view, Tablet tablet,
String clusterId) {
+ Long tabletId = tablet.getId();
+ for (Replica replica : tablet.getReplicas()) {
+ long beId = futureRouteBe(tabletId, (CloudReplica) replica,
clusterId);
+ if (beId != -1L) {
+ view.computeIfAbsent(beId, ignored -> new
HashSet<>()).add(tabletId);
+ }
+ }
+ }
+
+ private Map<Long, Map<Long, Set<Long>>> buildPartitionViews(Long
partitionId, String clusterId) {
+ Long dbId = partitionIdToDbId.get(partitionId);
+ Long tableId = partitionIdToTableId.get(partitionId);
+ Preconditions.checkNotNull(dbId);
+ Preconditions.checkNotNull(tableId);
+ Database db = Env.getCurrentInternalCatalog().getDbNullable(dbId);
+ if (db == null) {
+ return Collections.emptyMap();
+ }
+ OlapTable table = (OlapTable) db.getTableNullable(tableId);
+ if (table == null) {
+ return Collections.emptyMap();
+ }
+ Map<Long, Map<Long, Set<Long>>> views = new TreeMap<>();
+ table.readLock();
+ try {
+ if (db.getTableNullable(tableId) != table) {
+ return Collections.emptyMap();
+ }
+ Partition partition = table.getPartition(partitionId);
+ if (partition == null) {
+ return Collections.emptyMap();
+ }
+ for (MaterializedIndex index :
partition.getMaterializedIndices(IndexExtState.VISIBLE, true)) {
+ Map<Long, Set<Long>> view = new HashMap<>();
+ for (Tablet tablet : index.getTablets()) {
Review Comment:
[P3] Count the newly added route traversals in the scan metric
`currentRoundTabletScanCount` is incremented only by `loopCloudReplica`, so
this new traversal, `buildTableView`, and `countRoutes` are absent from
`cloud_tablet_rebalancer_tablet_scan_total`, which is documented as the total
route entries scanned per round. With one compute group and N already balanced
non-colocated tablets, the default round now performs 4N visits (route
completion, statistics, partition views, table views), but still reports 2N;
conflict-cache fills and rollback rebuilds add more unreported visits. That
makes the scan metric misleading when comparing the CPU/allocation cost of this
optimization. Include these new traversals in the round counter, including
rebuilds, or explicitly redefine the metric's scope.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
