github-actions[bot] commented on code in PR #68995:
URL: https://github.com/apache/doris/pull/68995#discussion_r4266185498
##########
fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudTabletRebalancer.java:
##########
@@ -1437,138 +1417,191 @@ 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);
+ }
+ }
+ }
+
+ 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;
Review Comment:
[P2] Resolve inflight state before reading the primary fallback
The warmup checker can finish A→B after `currentRouteBe()` returns A but
before the inflight lookup: it publishes primary B and removes the task, so
this helper returns obsolete A. The checker and view builder both use table
read locks, which permit this overlap. Queuing cancellation does not serialize
successful completion.
For example, on A/B/C use a seven-bucket partition with base-index FUTURE
counts (3,2,2), three rollups with (3,2,2), and two with (2,3,2). PARTITION
skips every balanced index. If its A→B tablet completes during TABLE's read
above, the TABLE view becomes (17,13,12) instead of (16,14,12). Selecting that
tablet from A schedules A→C because fresh conflict counts see C=2 < A=3; the
tablet is actually at B, so its index becomes (3,1,3), requiring corrective
balancing. The resident cache also decrements the wrong source. Previously
partition/table/conflict membership came from the same snapshot, preventing
this particular disagreement.
Read the inflight task before the primary fallback, or revalidate the
effective source before scheduling, and add a deterministic
completion-during-view test.
--
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]
