You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@doris.apache.org by "deardeng (via GitHub)" <gi...@apache.org> on 2026/10/13 03:46:11 UTC
[PR] [improvement](fe) Reduce cloud balance round time and transient heap use [doris]
deardeng opened a new pull request, #68995:
URL: https://github.com/apache/doris/pull/68995
Problem Summary: Cloud tablet rebalancer rounds retained current and future BE-to-tablet indexes at partition and table levels as well as the global indexes, duplicating route membership across a round. Build partition and table route views when each balance phase visits them, retain only per-table counts during the global phase, and add a unit test for the index lifetime. Preserve global routes, warmup rollback, scheduling order, and transfer thresholds. Views built later in a round can observe concurrent route updates.
For an offline A/B test on the diandian metadata image, compile separate full FE packages from the pre-L2 and L2 source with Maven build cache disabled. Run fresh FE processes in A-B-B-A order with the same image, JDK 17.0.2, 128 GiB heap, configuration, and offline-startup patches. After 300 seconds of warmup, record about 360 seconds of JFR and GC logs per run. Excluding each first round, the two per-run median balance times are 159.273/168.325 seconds before and 106.040/107.717 seconds after L2, a 34.75% reduction using the means of those medians.
In separate A-B-B-A runs, trigger live class histograms at the same partition-balance stage after route statistics are built. Live heap is 35.511/35.512 GiB before and 16.322/16.321 GiB after L2, a 19.190 GiB (54.04%) reduction. The difference is almost entirely ConcurrentHashMap route indexes. After a round completes, live heap is about 16.05-16.09 GiB in both versions because those temporary indexes have been released.
Across the fixed JFR windows, Young GC pause totals are 8.096/7.651 seconds before and 3.584/3.341 seconds after L2; Young GC counts are 19/19 and 18/18. Rebalancer allocation per second does not materially decrease. Maximum total heap occupied before GC over each full run is 104.983/105.215 GiB before and 99.195/100.434 GiB after L2; this includes reclaimable objects and is not live heap. The offline FE had no real MS/BE or user query traffic, so these results do not establish production memory requirements or route migration throughput.
### Release note
None
### Check List (For Author)
- Test: Unit Test: 62 related FE tests passed. Build: both full FE packages passed `./build.sh --fe --clean` with Checkstyle and Maven build cache disabled. Manual test: the A-B-B-A JFR/GC test and separate phase-aligned live-heap A-B-B-A test described above. Regression test: cloud balance Docker cases passed in the author's local run.
- Behavior changed: Yes (partition and table route views are rebuilt when visited and may observe a concurrent route update within the same round)
- Does this need documentation: No
### What problem does this PR solve?
Issue Number: close #xxx
Related PR: #xxx
Problem Summary:
### Release note
None
### Check List (For Author)
- Test <!-- At least one of them must be included. -->
- [ ] Regression test
- [x] Unit Test
- [ ] Manual test (add detailed scripts or steps below)
- [ ] No need to test or manual test. Explain why:
- [ ] This is a refactor/code format and no logic has been changed.
- [ ] Previous test can cover this change.
- [ ] No code files have been changed.
- [ ] Other reason <!-- Add your reason? -->
- Behavior changed:
- [x] No.
- [ ] Yes. <!-- Explain the behavior change -->
- Does this need documentation?
- [x] No.
- [ ] Yes. <!-- Add document PR link here. eg: https://github.com/apache/doris-website/pull/1214 -->
### Check List (For Reviewer who merge this PR)
- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label <!-- Add branch pick label that this PR should merge into -->
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [improvement](fe) Reduce cloud balance round time and transient heap use [doris]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on code in PR #68995:
URL: https://github.com/apache/doris/pull/68995#discussion_r4266624323
##########
fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudTabletRebalancer.java:
##########
@@ -1775,48 +1805,48 @@ 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;
+ // A live view can be newer than the round-global route snapshot.
+ for (Long be : Preconditions.checkNotNull(clusterToBes.get(clusterId))) {
+ Set<Long> tablets = globalBeToTablets.get(be);
+ if (tablets != null) {
+ tablets.remove(tabletId);
+ }
+ }
}
- ConcurrentHashMap<Long, Set<Long>> tableBeMap = beToTabletsInTable.get(tableId);
- if (tableBeMap == null) {
- return;
+ if (destBe != null) {
+ fillGlobalBeToTablets(destBe, tabletId, DEFAULT_GLOBAL_TABLET_SET_FACTORY, globalBeToTablets);
}
- Set<Long> tableSrcTablets = tableBeMap.get(srcBe);
- if (tableSrcTablets != null) {
- tableSrcTablets.remove(tabletId);
+ if (clusterId.equals(tableCountClusterId)) {
+ Map<Long, Integer> counts = (variant == RouteVariant.CURRENT ? currentTableCounts : futureTableCounts)
+ .get(tableId);
+ if (counts != null) {
+ counts.merge(srcBe, -1, Integer::sum);
+ if (destBe != null) {
+ 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)) {
+ Map<Long, Integer> counts = (variant == RouteVariant.CURRENT ? currentIndexCounts : futureIndexCounts)
+ .get(Pair.of(tabletMeta.getPartitionId(), tabletMeta.getIndexId()));
+ if (counts != null) {
+ counts.merge(srcBe, -1, Integer::sum);
Review Comment:
[P2] Derive CURRENT cache deltas from the actual primary
After a switch from `async_warmup` to `peer_read_async_warmup`, the checker retains old tasks, so a tablet primary on A but warming to B can be picked from FUTURE B for a direct B-to-C transfer. This CURRENT update decrements B although the actual current source is A. Reconciling global membership does not make that delta correct.
With partition balancing disabled and `cloud_rebalance_number_threshold=1`, use four BEs A/B/C/D and one table whose partitions have CURRENT counts `(4,3,2,0)`, `(1,2,0,0)`, `(1,0,0,4)`. Let one A tablet in the first partition be warming to B; TABLE FUTURE totals are `(5,6,2,4)` and the default budget permits four attempts. Choose B-to-C and A-to-C in the target, B-to-A in the second partition, then A-to-B in the target. Before the fourth move, new CURRENT target counts are `(3,2,4,0)` but actual counts are `(2,3,4,0)`; the old current graph is `(3,3,3,0)` and rejects that move. The new check permits it, worsening actual placement to `(1,4,4,0)`.
Derive CURRENT deltas from the actual primary, or rebuild/invalidate that cache when the selected FUTURE source differs; apply this to table counts too. Add a residual-task mode-change test. This instance uses an uncanceled task and CURRENT counts, separate from the earlier FUTURE cancellation issue.
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [improvement](fe) Reduce cloud balance round time and transient heap use [doris]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [improvement](fe) Reduce cloud balance round time and transient heap use [doris]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #68995:
URL: https://github.com/apache/doris/pull/68995#issuecomment-6261523894
Codex automated review failed and did not complete.
Error: Review recovery stopped: No final review submission was declared
Workflow run: https://github.com/apache/doris/actions/runs/39210680485
Please inspect the workflow logs and rerun the review after the underlying issue is resolved.
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [improvement](fe) Reduce cloud balance round time and transient heap use [doris]
Posted by "hello-stephen (via GitHub)" <gi...@apache.org>.
hello-stephen commented on PR #68995:
URL: https://github.com/apache/doris/pull/68995#issuecomment-6137763236
Thank you for your contribution to Apache Doris.
Don't know what should be done next? See [How to process your PR](https://doris.apache.org/community/how-to-contribute/review-and-ci).
Please clearly describe your PR:
1. What problem was fixed (it's best to include specific error reporting information). How it was fixed.
2. Which behaviors were modified. What was the previous behavior, what is it now, why was it modified, and what possible impacts might there be.
3. What features were added. Why was this function added?
4. Which code was refactored and why was this part of the code refactored?
5. Which functions were optimized and what is the difference before and after the optimization?
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [improvement](fe) Reduce cloud balance round time and transient heap use [doris]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on code in PR #68995:
URL: https://github.com/apache/doris/pull/68995#discussion_r4254639327
##########
fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudTabletRebalancer.java:
##########
@@ -1983,69 +1976,72 @@ 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;
+ }
+ Map<Long, Integer> indexCounts = new HashMap<>();
+ countRoutes(indexCounts, index, clusterId, variant);
Review Comment:
[P2] Cache index counts instead of recounting every candidate
Every TABLE/GLOBAL attempt now scans the entire index under the table read lock, including candidates rejected immediately afterward. For example, with the bucket limit raised, 100 BEs can have a balanced 100,000-tablet index (1,000 per BE) plus 100 singleton tables still routed to one BE after expansion. The partition/table phases skip these distributions, but GLOBAL sees 1,100 versus 1,000 tablets and tries 100 candidates with the default budget. Most candidates belong to the balanced index and are rejected, adding roughly 9 million route visits to the existing roughly 110,000 source-set sampling visits. The table-count cache cannot help because the index check returns first. Cache compact per-index counts and maintain/invalidate them with route changes to preserve the memory saving without repeatedly recounting an unchanged index.
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [improvement](fe) Reduce cloud balance round time and transient heap use [doris]
Posted by "deardeng (via GitHub)" <gi...@apache.org>.
deardeng commented on PR #68995:
URL: https://github.com/apache/doris/pull/68995#issuecomment-6139411610
/review
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [improvement](fe) Reduce cloud balance round time and transient heap use [doris]
Posted by "deardeng (via GitHub)" <gi...@apache.org>.
deardeng commented on PR #68995:
URL: https://github.com/apache/doris/pull/68995#issuecomment-6160221901
/review
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [improvement](fe) Reduce cloud balance round time and transient heap use [doris]
Posted by "deardeng (via GitHub)" <gi...@apache.org>.
deardeng commented on PR #68995:
URL: https://github.com/apache/doris/pull/68995#issuecomment-6137766370
/review
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [improvement](fe) Reduce cloud balance round time and transient heap use [doris]
Posted by "deardeng (via GitHub)" <gi...@apache.org>.
deardeng closed pull request #68995: [improvement](fe) Reduce cloud balance round time and transient heap use
URL: https://github.com/apache/doris/pull/68995
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [improvement](fe) Reduce cloud balance round time and transient heap use [doris]
Posted by "hello-stephen (via GitHub)" <gi...@apache.org>.
hello-stephen commented on PR #68995:
URL: https://github.com/apache/doris/pull/68995#issuecomment-6139271988
<details>
<summary>ClickBench: <b>Total hot run time: 24.69 s</b></summary>
```
machine: 'aliyun_ecs.c7a.8xlarge_32C64G'
scripts: https://github.com/apache/doris/tree/master/tools/clickbench-tools
ClickBench test result on commit b56c451e9b1167cbab5c105bf278ef0fa5a0ae26, data reload: false
query1 0.01 0.01 0.00
query2 0.10 0.05 0.05
query3 0.26 0.11 0.13
query4 1.61 0.13 0.12
query5 0.26 0.25 0.24
query6 1.17 0.69 0.68
query7 0.04 0.01 0.00
query8 0.06 0.04 0.04
query9 0.41 0.35 0.36
query10 0.56 0.58 0.55
query11 0.20 0.14 0.13
query12 0.18 0.14 0.14
query13 0.49 0.50 0.50
query14 0.99 0.96 0.96
query15 0.65 0.63 0.61
query16 0.34 0.36 0.35
query17 1.13 1.13 1.17
query18 0.23 0.21 0.21
query19 2.11 1.94 2.00
query20 0.02 0.02 0.01
query21 15.44 0.29 0.14
query22 4.45 0.06 0.05
query23 16.27 0.32 0.13
query24 2.92 0.59 0.56
query25 0.11 0.07 0.07
query26 1.58 0.24 0.14
query27 0.06 0.04 0.04
query28 2.34 0.62 0.41
query29 12.52 4.19 3.32
query30 0.28 0.14 0.15
query31 2.77 0.57 0.33
query32 3.24 0.60 0.52
query33 3.16 3.24 3.19
query34 15.46 4.01 3.41
query35 3.34 3.33 3.34
query36 0.54 0.43 0.42
query37 0.10 0.06 0.06
query38 0.06 0.04 0.04
query39 0.04 0.02 0.02
query40 0.16 0.15 0.14
query41 0.08 0.03 0.03
query42 0.03 0.04 0.04
query43 0.04 0.04 0.04
Total cold run time: 95.81 s
Total hot run time: 24.69 s
```
</details>
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [improvement](fe) Reduce cloud balance round time and transient heap use [doris]
Posted by "deardeng (via GitHub)" <gi...@apache.org>.
deardeng commented on PR #68995:
URL: https://github.com/apache/doris/pull/68995#issuecomment-6188916217
/review
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [improvement](fe) Reduce cloud balance round time and transient heap use [doris]
Posted by "hello-stephen (via GitHub)" <gi...@apache.org>.
hello-stephen commented on PR #68995:
URL: https://github.com/apache/doris/pull/68995#issuecomment-6140223601
# FE UT Coverage Report
Increment line coverage ` 51.22% (105/205)` :tada:
[Increment coverage report](http://coverage.selectdb-in.cc/coverage/68995_b56c451e9b1167cbab5c105bf278ef0fa5a0ae26/fe_increment_report/index.html)
[Complete coverage report](http://coverage.selectdb-in.cc/coverage/68995_b56c451e9b1167cbab5c105bf278ef0fa5a0ae26/fe_report/index.html)
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [improvement](fe) Reduce cloud balance round time and transient heap use [doris]
Posted by "deardeng (via GitHub)" <gi...@apache.org>.
deardeng commented on PR #68995:
URL: https://github.com/apache/doris/pull/68995#issuecomment-6142982784
/review
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [improvement](fe) Reduce cloud balance round time and transient heap use [doris]
Posted by "deardeng (via GitHub)" <gi...@apache.org>.
deardeng commented on PR #68995:
URL: https://github.com/apache/doris/pull/68995#issuecomment-6206748061
/review
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [improvement](fe) Reduce cloud balance round time and transient heap use [doris]
Posted by "deardeng (via GitHub)" <gi...@apache.org>.
deardeng commented on code in PR #68995:
URL: https://github.com/apache/doris/pull/68995#discussion_r4293074686
##########
fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudTabletRebalancer.java:
##########
@@ -1302,16 +1266,39 @@ private void handleWarmupBatchFailure(List<WarmupTabletTask> tasks, Exception e)
}
private void revertWarmupState(WarmupTabletTask task) {
- updateBeToTablets(task.pickedTabletId, task.destBe, task.srcBe,
- futureBeToTabletsGlobal, futureBeToTabletsInTable, futurePartitionToTablets);
- tabletToInfightTask.remove(new InfightTablet(task.pickedTabletId, task.clusterId));
+ InfightTablet key = new InfightTablet(task.pickedTabletId, task.clusterId);
+ InfightTask inflight = tabletToInfightTask.get(key);
+ if (inflight == null || inflight.srcBe != task.srcBe || inflight.destBe != task.destBe
+ || !tabletToInfightTask.remove(key, inflight)) {
Review Comment:
Thanks for the detailed interleaving. I agree that the checker can retain an InfightTask after the daemon removes it. That task ownership race predates this PR, although lazy views may turn it into an extra migration in the scenario described.
The finding is based on static analysis; we do not yet have a runtime reproduction or evidence of data loss. Serializing completion, rollback, and the other task removal paths would expand this performance PR substantially and introduce its own concurrency risk. I propose to keep this PR scoped and track this race separately, with a deterministic reproducer before changing the task lifecycle.
This reply accepts the possible migration churn; it does not claim the race is fixed.
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [improvement](fe) Reduce cloud balance round time and transient heap use [doris]
Posted by "deardeng (via GitHub)" <gi...@apache.org>.
deardeng commented on PR #68995:
URL: https://github.com/apache/doris/pull/68995#issuecomment-6261219183
/review
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [improvement](fe) Reduce cloud balance round time and transient heap use [doris]
Posted by "hello-stephen (via GitHub)" <gi...@apache.org>.
hello-stephen commented on PR #68995:
URL: https://github.com/apache/doris/pull/68995#issuecomment-6139091435
<details>
<summary>TPC-H: <b>Total hot run time: 28963 ms</b></summary>
```
machine: 'aliyun_ecs.c7a.8xlarge_32C64G'
scripts: https://github.com/apache/doris/tree/master/tools/tpch-tools
Tpch sf100 test result on commit b56c451e9b1167cbab5c105bf278ef0fa5a0ae26, data reload: false
------ Round 1 ----------------------------------
============================================
q1 17575 5651 5519 5519
q2 2108 298 273 273
q3 10244 1370 836 836
q4 4672 485 348 348
q5 7413 801 548 548
q6 189 186 155 155
q7 772 801 609 609
q8 9257 1321 1048 1048
q9 5648 4433 4412 4412
q10 6795 1271 1006 1006
q11 435 254 234 234
q12 638 409 286 286
q13 18053 3071 2379 2379
q14 282 275 246 246
q15 q16 724 735 677 677
q17 1170 746 570 570
q18 6999 6204 6175 6175
q19 1104 977 618 618
q20 392 335 226 226
q21 4895 2513 2583 2513
q22 425 333 285 285
Total cold run time: 99790 ms
Total hot run time: 28963 ms
----- Round 2, with runtime_filter_mode=off -----
============================================
q1 6503 6417 6450 6417
q2 753 604 567 567
q3 5477 5683 5228 5228
q4 2012 2140 1583 1583
q5 5897 5716 5703 5703
q6 241 189 160 160
q7 2305 2030 1813 1813
q8 3094 2795 2765 2765
q9 8170 7981 7716 7716
q10 3652 3625 3237 3237
q11 586 409 375 375
q12 672 692 500 500
q13 2748 3073 2383 2383
q14 279 297 269 269
q15 q16 658 696 606 606
q17 7700 6979 6954 6954
q18 13144 12256 13070 12256
q19 941 794 819 794
q20 2184 2159 1945 1945
q21 5765 4714 4892 4714
q22 529 452 422 422
Total cold run time: 73310 ms
Total hot run time: 66407 ms
```
</details>
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [improvement](fe) Reduce cloud balance round time and transient heap use [doris]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [improvement](fe) Reduce cloud balance round time and transient heap use [doris]
Posted by "deardeng (via GitHub)" <gi...@apache.org>.
deardeng commented on PR #68995:
URL: https://github.com/apache/doris/pull/68995#issuecomment-6159446265
/review
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [improvement](fe) Reduce cloud balance round time and transient heap use [doris]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on code in PR #68995:
URL: https://github.com/apache/doris/pull/68995#discussion_r4257418473
##########
fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudTabletRebalancer.java:
##########
@@ -1775,48 +1801,42 @@ 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;
+ // A live view can be newer than the round-global route snapshot.
+ for (Long be : Preconditions.checkNotNull(clusterToBes.get(clusterId))) {
Review Comment:
[P2] Skip obsolete rollbacks for a removed compute group
When a compute group is dropped while it has inflight warmups, `CloudClusterChecker` removes all of its backends, so the next `buildClusterToBackendMap`/`statRouteInfo` omit that group. The independent warmup checker queues its tasks for cancellation, but draining each task reaches this fallback with no global source membership and `clusterToBes.get(clusterId) == null`. `checkNotNull` then throws after only that task has been removed, aborting the round before `performBalancing`. With N queued warmups, remaining compute groups lose N balance rounds, each repeating the full route/statistics scans. Treat cancellation for a group absent from the round map as obsolete cleanup, without restoring membership for its removed backends, and drain all such tasks in one round. Add a test that removes an entire group with multiple inflight tasks while another group remains live.
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [improvement](fe) Reduce cloud balance round time and transient heap use [doris]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #68995:
URL: https://github.com/apache/doris/pull/68995#issuecomment-6263921240
Codex automated review failed and did not complete.
Error: Review recovery stopped: No final review submission was declared
Workflow run: https://github.com/apache/doris/actions/runs/39227434743
Please inspect the workflow logs and rerun the review after the underlying issue is resolved.
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [improvement](fe) Reduce cloud balance round time and transient heap use [doris]
Posted by "deardeng (via GitHub)" <gi...@apache.org>.
deardeng commented on PR #68995:
URL: https://github.com/apache/doris/pull/68995#issuecomment-6263633288
/review
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [improvement](fe) Reduce cloud balance round time and transient heap use [doris]
Posted by "deardeng (via GitHub)" <gi...@apache.org>.
deardeng commented on PR #68995:
URL: https://github.com/apache/doris/pull/68995#issuecomment-6137765632
run buildall
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [improvement](fe) Reduce cloud balance round time and transient heap use [doris]
Posted by "hello-stephen (via GitHub)" <gi...@apache.org>.
hello-stephen commented on PR #68995:
URL: https://github.com/apache/doris/pull/68995#issuecomment-6139211093
<details>
<summary>TPC-DS: <b>Total hot run time: 151622 ms</b></summary>
```
machine: 'aliyun_ecs.c7a.8xlarge_32C64G'
scripts: https://github.com/apache/doris/tree/master/tools/tpcds-tools
TPC-DS sf100 test result on commit b56c451e9b1167cbab5c105bf278ef0fa5a0ae26, data reload: false
query5 4311 581 450 450
query6 438 205 194 194
query7 4806 437 223 223
query8 329 181 167 167
query9 8794 4031 4012 4012
query10 467 324 269 269
query11 5984 3442 3191 3191
query12 143 91 84 84
query13 1256 450 322 322
query14 6558 4769 4485 4485
query14_1 4286 4222 4237 4222
query15 215 205 184 184
query16 967 453 425 425
query17 862 655 541 541
query18 2423 420 309 309
query19 200 167 132 132
query20 84 81 80 80
query21 214 134 112 112
query22 12941 13084 12799 12799
query23 13114 12404 12066 12066
query23_1 12349 12134 12334 12134
query24 7114 823 461 461
query24_1 475 469 489 469
query25 556 427 366 366
query26 1264 266 148 148
query27 2735 442 271 271
query28 4643 1953 1969 1953
query29 1568 600 461 461
query30 298 223 183 183
query31 887 772 647 647
query32 148 97 93 93
query33 515 303 250 250
query34 986 850 492 492
query35 743 780 648 648
query36 824 809 743 743
query37 135 100 91 91
query38 1782 1759 1679 1679
query39 702 728 681 681
query39_1 657 689 684 684
query40 222 117 97 97
query41 76 70 69 69
query42 85 87 83 83
query43 371 384 331 331
query44 1312 722 734 722
query45 179 179 164 164
query46 859 953 563 563
query47 2865 3032 2797 2797
query48 297 302 206 206
query49 559 399 276 276
query50 683 275 217 217
query51 10304 10301 10349 10301
query52 78 78 73 73
query53 195 205 151 151
query54 232 204 176 176
query55 72 66 62 62
query56 231 201 196 196
query57 1595 1611 1541 1541
query58 268 260 247 247
query59 2370 2382 2134 2134
query60 271 235 221 221
query61 150 150 136 136
query62 387 338 285 285
query63 196 157 164 157
query64 2744 922 742 742
query65 3444 3360 3410 3360
query66 1784 408 295 295
query67 20207 20312 20070 20070
query68 2849 1004 671 671
query69 385 276 242 242
query70 922 845 835 835
query71 310 235 220 220
query72 2661 2625 2280 2280
query73 511 524 290 290
query74 4599 4473 4316 4316
query75 2324 2319 1915 1915
query76 2281 998 622 622
query77 354 399 292 292
query78 9068 9053 8448 8448
query79 996 858 513 513
query80 1238 427 331 331
query81 586 325 280 280
query82 611 135 109 109
query83 303 190 176 176
query84 317 126 102 102
query85 849 434 373 373
query86 401 248 236 236
query87 2016 1962 1829 1829
query88 3596 2707 2679 2679
query89 343 291 256 256
query90 1917 177 180 177
query91 158 150 122 122
query92 96 87 91 87
query93 1050 992 571 571
query94 654 288 249 249
query95 600 348 323 323
query96 669 519 231 231
query97 2482 2425 2326 2326
query98 159 148 141 141
query99 781 761 655 655
Total cold run time: 234610 ms
Total hot run time: 151622 ms
```
</details>
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [improvement](fe) Reduce cloud balance round time and transient heap use [doris]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on code in PR #68995:
URL: https://github.com/apache/doris/pull/68995#discussion_r4290370668
##########
fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudTabletRebalancer.java:
##########
@@ -1302,16 +1266,39 @@ private void handleWarmupBatchFailure(List<WarmupTabletTask> tasks, Exception e)
}
private void revertWarmupState(WarmupTabletTask task) {
- updateBeToTablets(task.pickedTabletId, task.destBe, task.srcBe,
- futureBeToTabletsGlobal, futureBeToTabletsInTable, futurePartitionToTablets);
- tabletToInfightTask.remove(new InfightTablet(task.pickedTabletId, task.clusterId));
+ InfightTablet key = new InfightTablet(task.pickedTabletId, task.clusterId);
+ InfightTask inflight = tabletToInfightTask.get(key);
+ if (inflight == null || inflight.srcBe != task.srcBe || inflight.destBe != task.destBe
+ || !tabletToInfightTask.remove(key, inflight)) {
Review Comment:
[P2] Serialize rollback with already-started warmup completion
Removing the task here does not revoke a checker that already obtained its `InfightTask` at lines 923-925: `handleWarmupCompletion` can still publish its destination at line 1850 after this rollback restores GLOBAL to the old primary. For example, a warmup RPC accepted by B can time out at FE while the checker reports it done. Pause that checker before publication, drain the failure here, then resume it before the live partition/table views are built.
With healthy A/B/C, let the first seven-tablet table have CURRENT `(4,1,2)` and FUTURE `(3,2,2)` through one A-to-B task, and a second single-index table have `(2,3,2)`. Rollback leaves GLOBAL `(6,4,4)`; the held completion makes the actual tables `(3,2,2)` and `(2,3,2)`, so both live phases skip. GLOBAL can then pick the raced tablet from A for A-to-C: fresh index/table counts allow `A=3 > C=2`, although the tablet is actually at B. Its index becomes `(3,1,3)`. The old retained partition graph saw `(4,1,2)`, scheduled partition correction, and gated GLOBAL.
This is a separate post-snapshot rollback producer from the fixed primary/task lookup threads. Coordinate task ownership and primary publication between completion and rollback (or process both on the daemon), and add a latch test for this held-callback interleaving.
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [improvement](fe) Reduce cloud balance round time and transient heap use [doris]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org