This is an automated email from the ASF dual-hosted git repository.
Jackie-Jiang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pinot.git
The following commit(s) were added to refs/heads/master by this push:
new 867b53b7489 Skip full target recompute on IdealState version change
for strict realtime rebalance that only moves tier segments (#19054)
867b53b7489 is described below
commit 867b53b7489d476407a09ca5f40b12a42693e58c
Author: Xiaotian (Jackie) Jiang <[email protected]>
AuthorDate: Thu Jul 23 13:50:58 2026 -0700
Skip full target recompute on IdealState version change for strict realtime
rebalance that only moves tier segments (#19054)
---
.../helix/core/rebalance/TableRebalancer.java | 41 ++++++++++++++++++++--
1 file changed, 38 insertions(+), 3 deletions(-)
diff --git
a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/rebalance/TableRebalancer.java
b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/rebalance/TableRebalancer.java
index 643a4896e4e..63490ca54a6 100644
---
a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/rebalance/TableRebalancer.java
+++
b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/rebalance/TableRebalancer.java
@@ -42,6 +42,7 @@ import java.util.function.ToIntFunction;
import java.util.stream.Collectors;
import javax.annotation.Nullable;
import org.apache.commons.collections4.CollectionUtils;
+import org.apache.commons.collections4.MapUtils;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.commons.lang3.tuple.Triple;
import org.apache.helix.AccessOption;
@@ -653,11 +654,22 @@ public class TableRebalancer {
// If all the segments to be moved remain unchanged (same instance
state map) in the new ideal state, apply
// the same target instance state map for these segments to the new
ideal state as the target assignment
segmentsToMoveChanged = false;
- if (segmentAssignment instanceof
BaseStrictRealtimeSegmentAssignment) {
- // For StrictRealtimeSegmentAssignment, we need to recompute the
target assignment because the assignment
- // for new added segments is based on the existing assignment
+ if (isStrictRealtimeSegmentAssignment
+ && !isMovingOnlyTierSegments(segmentsToMove,
providedTierToSegmentsMap)) {
+ // For strict segment assignment, a newly added segment (a new
consuming segment or an uploaded segment) is
+ // assigned by the instance partitions, then overridden to match
the existing assignment of its partition
+ // in the IdealState when they differ, keeping the partition
collocated. This rebalance moves a non-tier
+ // segment, i.e. the base placement of a partition, so a segment
added to the IdealState while we wait can
+ // be placed on the partition's old placement and must be
re-collocated to the target: recompute the full
+ // target assignment.
segmentsToMoveChanged = true;
} else {
+ // Non-strict segment assignment, or strict assignment that only
moves tier segments (completed segments
+ // assigned to a tier). In the latter case the base placements do
not move, so a segment added while we
+ // wait is placed consistently with the target; a full recompute
is only needed if a segment we are moving
+ // actually changed state. Skipping it (a full recompute reads
segment ZK metadata for the completed
+ // segments) narrows the window in which the versioned IdealState
update below can lose the race to a
+ // concurrent write on a continuously-ingesting table.
for (String segment : segmentsToMove) {
Map<String, String> oldInstanceStateMap =
oldAssignment.get(segment);
Map<String, String> currentInstanceStateMap =
currentAssignment.get(segment);
@@ -2228,6 +2240,29 @@ public class TableRebalancer {
}
}
+ /// Returns whether all the segments to be moved are tier segments, i.e.
contained in the provided tier-to-segments
+ /// map. When true, this rebalance only relocates completed segments onto
tiers and leaves the base (non-tier)
+ /// placements of the partitions untouched, so a segment added to the
IdealState mid-rebalance is still placed
+ /// consistently with the target. Returns false when the tier segments are
not pre-computed (the map is null/empty),
+ /// conservatively treating the moves as base placement changes.
+ private static boolean isMovingOnlyTierSegments(List<String> segmentsToMove,
+ @Nullable Map<String, Set<String>> providedTierToSegmentsMap) {
+ if (segmentsToMove.isEmpty()) {
+ return true;
+ }
+ if (MapUtils.isEmpty(providedTierToSegmentsMap)) {
+ return false;
+ }
+ Set<String> tierSegments;
+ if (providedTierToSegmentsMap.size() == 1) {
+ tierSegments = providedTierToSegmentsMap.values().iterator().next();
+ } else {
+ tierSegments = new HashSet<>();
+ providedTierToSegmentsMap.values().forEach(tierSegments::addAll);
+ }
+ return tierSegments.size() >= segmentsToMove.size() &&
tierSegments.containsAll(segmentsToMove);
+ }
+
@VisibleForTesting
static Set<String> getMovingConsumingSegments(Map<String, Map<String,
String>> currentAssignment,
Map<String, Map<String, String>> targetAssignment) {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
