vagetablechicken commented on a change in pull request #5010: URL: https://github.com/apache/incubator-doris/pull/5010#discussion_r549560214
########## File path: fe/fe-core/src/main/java/org/apache/doris/clone/TwoDimensionalGreedyRebalanceAlgo.java ########## @@ -0,0 +1,329 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package org.apache.doris.clone; + +import com.google.common.base.Objects; +import com.google.common.base.Preconditions; +import com.google.common.collect.Lists; +import com.google.common.collect.Sets; +import com.google.common.collect.TreeMultimap; +import org.apache.doris.catalog.TabletInvertedIndex.PartitionBalanceInfo; +import org.apache.doris.clone.PartitionRebalancer.ClusterBalanceInfo; +import org.apache.doris.common.Pair; +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; + +import java.util.List; +import java.util.NavigableSet; +import java.util.Random; +import java.util.Set; +import java.util.stream.Collectors; + +/* + * A two-dimensional greedy rebalancing algorithm. The two dims are cluster and partition. It'll generate multiple `PartitionMove`, + * only decide which partition to move, fromBe, toBe. The next step is to select a tablet to move. + * + * From among moves that decrease the skew of a most skewed partition, it prefers ones that reduce the skew of the cluster. + * A cluster is considered balanced when the skew of every partition is <= 1 and the skew of the cluster is <= 1. + * The skew of the cluster is defined as the difference between the maximum total replica count over all bes and the + * minimum total replica count over all bes. + * + * This class is modified from kudu TwoDimensionalGreedyAlgo. + */ +public class TwoDimensionalGreedyRebalanceAlgo { + private static final Logger LOG = LogManager.getLogger(TwoDimensionalGreedyRebalanceAlgo.class); + + private final EqualSkewOption equalSkewOption; + private static final Random rand = new Random(System.currentTimeMillis()); + + public static class PartitionMove { + Long partitionId; + Long indexId; + Long fromBe; + Long toBe; + + PartitionMove(Long p, Long i, Long f, Long t) { + this.partitionId = p; + this.indexId = i; + this.fromBe = f; + this.toBe = t; + } + + @Override + public boolean equals(Object o) { + if (this == o) return true; + if (o == null || getClass() != o.getClass()) return false; + PartitionMove that = (PartitionMove) o; + return Objects.equal(partitionId, that.partitionId) && + Objects.equal(indexId, that.indexId) && + Objects.equal(fromBe, that.fromBe) && + Objects.equal(toBe, that.toBe); + } + + @Override + public int hashCode() { + return Objects.hashCode(partitionId, indexId, fromBe, toBe); + } + + @Override + public String toString() { + return "ReplicaMove{" + + "pid=" + partitionId + "-" + indexId + + ", from=" + fromBe + + ", to=" + toBe + + '}'; + } + } + + public enum EqualSkewOption { + // generally only be used on unit test + PICK_FIRST, + PICK_RANDOM + } + + public enum ExtremumType { + MAX, + MIN + } + + public static class IntersectionResult { + Long replicaCountPartition; + Long replicaCountTotal; + List<Long> beWithExtremumCount; + List<Long> intersection; + } + + TwoDimensionalGreedyRebalanceAlgo() { + this(EqualSkewOption.PICK_RANDOM); + } + + TwoDimensionalGreedyRebalanceAlgo(EqualSkewOption equalSkewOption) { + this.equalSkewOption = equalSkewOption; + } + + // maxMovesNum: Value of '0' is a shortcut for 'the possible maximum'. + // May modify the ClusterBalanceInfo + public List<PartitionMove> getNextMoves(ClusterBalanceInfo info, int maxMovesNum) { + Preconditions.checkArgument(maxMovesNum >= 0); + if (maxMovesNum == 0) { + maxMovesNum = Integer.MAX_VALUE; + } + + if (info.partitionInfoBySkew.isEmpty()) { + // Check for the consistency of the 'ClusterBalanceInfo' parameter: if no information is given on + // the partition skew, partition count for all the be should be 0. + // Keys are ordered by the natural ordering, so we can get the last(max) key to know if all keys are 0. + NavigableSet<Long> keySet = info.beByTotalReplicaCount.keySet(); + LOG.debug(keySet); + Preconditions.checkState(keySet.isEmpty() || keySet.last() == 0L, + "non-zero replica count on be while no partition skew information in skewMap"); + // Nothing to balance: cluster is empty. + return Lists.newArrayList(); + } + + List<PartitionMove> moves = Lists.newArrayList(); + for (int i = 0; i < maxMovesNum; ++i) { + PartitionMove move = getNextMove(info.beByTotalReplicaCount, info.partitionInfoBySkew); + if (move == null || !(applyMove(move, info.beByTotalReplicaCount, info.partitionInfoBySkew))) { + // 1. No replicas to move. + // 2. Apply to info failed, it's useless to get next move from the same info. + break; + } + moves.add(move); + } + + return moves; + } + + private PartitionMove getNextMove(TreeMultimap<Long, Long> beByTotalReplicaCount, + TreeMultimap<Long, PartitionBalanceInfo> skewMap) { + PartitionMove move = null; + if (skewMap.isEmpty() || beByTotalReplicaCount.isEmpty()) { + return null; + } + long maxPartitionSkew = skewMap.keySet().last(); + long maxBeSkew = beByTotalReplicaCount.keySet().last() - beByTotalReplicaCount.keySet().first(); + + // 1. Every partition is balanced(maxPartitionSkew<=1) and any move will unbalance a partition, so there + // is no potential for the greedy algorithm to balance the cluster. + // 2. Every partition is balanced(maxPartitionSkew<=1) and the cluster as a whole is balanced(maxBeSkew<=1). + if (maxPartitionSkew == 0L || (maxPartitionSkew <= 1L && maxBeSkew <= 1L)) { Review comment: The algo should get accurate calculation results. And I think there is no reason yet to relax the condition. ---------------------------------------------------------------- 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. For queries about this service, please contact Infrastructure at: us...@infra.apache.org --------------------------------------------------------------------- To unsubscribe, e-mail: commits-unsubscr...@doris.apache.org For additional commands, e-mail: commits-h...@doris.apache.org
