github-actions[bot] commented on code in PR #68995:
URL: https://github.com/apache/doris/pull/68995#discussion_r4281423788
##########
fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudTabletRebalancer.java:
##########
@@ -1413,13 +1396,10 @@ public void statRouteInfo() {
: 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, tableId, partitionId, indexId,
routeTabletId,
- futureGlobalTabletSetFactory,
- tmpFutureBeToTabletsGlobal,
futureBeToTabletsInTable, futurePartitionToTablets);
+ fillGlobalBeToTablets(beId, routeTabletId,
currentGlobalTabletSetFactory,
+ tmpBeToTabletsGlobal);
+ fillGlobalBeToTablets(futureBeId, routeTabletId,
futureGlobalTabletSetFactory,
Review Comment:
[P2] Resolve inflight state before retaining the GLOBAL route
The task-before-primary fix in `futureRouteBe` leaves a separate instance
here: `statRouteInfo` still reads primary A before looking up the inflight
task. If the checker completes A→B between those reads, this GLOBAL snapshot
retains A, while the new partition/table views see B. With healthy A/B/C and
two single-index tables whose actual counts are `(3,2,2)` and `(2,3,2)`, both
live phases skip balancing, but the stale GLOBAL counts are `(6,4,4)`. Under
the defaults, GLOBAL can select the raced tablet from A for A→C; fresh
index/table counts both allow it because A=3>C=2, although it is actually at B.
Its index becomes `(3,1,3)`. The old shared snapshot instead saw `(4,1,2)` at
partition scope and blocked GLOBAL after a partition move. Resolve the task
before the primary fallback in this builder too, or revalidate the effective
source before GLOBAL scheduling, and cover completion during `statRouteInfo` in
a deterministic test. This is the remaining GLOBAL-input location, dist
inct from the fixed TABLE-helper thread.
##########
fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudTabletRebalancer.java:
##########
@@ -1471,138 +1452,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);
Review Comment:
[P2] Invalidate only partition views affected by rollback
`processFailedWarmupTasks()` drains a global queue and returns true even for
an obsolete task or a rollback in another partition/compute group. Every such
result here rebuilds *all* index views of the current partition, scanning and
allocating all tablet memberships under its table read lock. If outstanding RPC
batches for other partitions keep failing between index visits, an otherwise
balanced partition with R indices and T tablets is rebuilt up to R times:
O(R*T) work (O(R²*N) for N tablets per index), instead of one O(T) build. This
can multiply round CPU/allocation and block DDL during warmup failures. Return
the affected index identities and refresh only a changed view, build each next
index independently, or apply rollback deltas; avoid invalidating this
partition for unrelated/no-op failures.
--
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]
