kangkaisen commented on a change in pull request #1139: Improve colocate table
balance logic for backend added
URL: https://github.com/apache/incubator-doris/pull/1139#discussion_r283083777
##########
File path: fe/src/main/java/org/apache/doris/clone/ColocateTableBalancer.java
##########
@@ -507,111 +508,133 @@ private Long
selectCloneBackendIdForRemove(com.google.common.collect.Table<Long,
}
/**
- * balance after new backend added
- *
- * 1 compute the the number of bucket seqs need to move from the each old
backend and
- * the number of bucket seqs need to move to the each new backend
- * 2 select the clone target Backend for the new Replica
- * 3 mark colocate group balancing
- * 4 add a Migration Job
- * 5 update the ColocateTableIndex's backendsPerBucketSeq
+ * 1 compute which bucket seq need to migrate, the migrate source backend,
the migrate target backend
+ * 2 mark colocate group balancing in colocate meta
+ * 3 update colcate backendsPerBucketSeq meta
*
* For example:
- * There are 3 backend and 4 tablet, and replicateNum is 3.
- *
- * the mapping from tablet to backend to is following:
- *
- * tablet1 : [1, 2, 3]
- * tablet2 : [2, 1, 3]
- * tablet3 : [3, 2, 1]
- * tablet4 : [1, 2, 3]
- *
- * After Adding a new backend:
- *
- * the needMoveBucketSeqs = 4 * 3 / (3 + 1) = 3
- * the bucketSeqsPerNewBackend = 3 / 1 = 1
*
- * After balancing, the mapping from tablet to backend to is following:
- *
- * tablet1 : [4, 2, 3]
- * tablet2 : [4, 1, 3]
- * tablet3 : [4, 2, 1]
- * tablet4 : [1, 2, 3]
+ * the old backendsPerBucketSeq is:
+ * [[1, 2, 3], [4, 1, 2], [3, 4, 1], [2, 3, 4], [1, 2, 3]]
+ *
+ * after we add two new backends: [5, 6]
+ *
+ * the balanced backendsPerBucketSeq will become:
+ * [[5, 6, 3], [6, 1, 2], [5, 4, 1], [2, 3, 4], [1, 2, 3]]
*
*/
private void balanceForBackendAdded(Long groupId, Database db, List<Long>
addedBackendIds) {
ColocateTableIndex colocateIndex = Catalog.getCurrentColocateIndex();
- com.google.common.collect.Table<Long, Integer, Long>
newGroup2BackendsPerBucketSeq = HashBasedTable.create();
-
List<List<Long>> backendsPerBucketSeq =
colocateIndex.getBackendsPerBucketSeq(groupId);
- int replicateNum = backendsPerBucketSeq.get(0).size();
- Set<Long> allGroupBackendIds =
colocateIndex.getBackendsByGroup(groupId);
-
- List<List<Long>> newBackendsPerBucketSeq =
deepCopy(backendsPerBucketSeq);
-
- int needMoveBucketSeqs = backendsPerBucketSeq.size() * replicateNum /
(allGroupBackendIds.size() + addedBackendIds.size());
- int bucketSeqsPerNewBackend = needMoveBucketSeqs /
addedBackendIds.size();
- LOG.info("for colocate group {}, needMoveBucketSeqs : {} ,
bucketSeqPerNewBackend: {}", groupId, needMoveBucketSeqs,
bucketSeqsPerNewBackend);
db.readLock();
try {
- List<Long> allTableIds = colocateIndex.getAllTableIds(groupId);
- for (long tableId : allTableIds) {
- OlapTable olapTable = (OlapTable) db.getTable(tableId);
- for (Partition partition : olapTable.getPartitions()) {
- replicateNum =
olapTable.getPartitionInfo().getReplicationNum(partition.getId());
- for (MaterializedIndex index :
partition.getMaterializedIndices()) {
- List<Tablet> tablets = index.getTablets();
- for (int i = 0; i < tablets.size() && i <
needMoveBucketSeqs; i++) {
- Tablet tablet = tablets.get(i);
- List<Replica> replicas = tablet.getReplicas();
- List<Long> sortedReplicaIds =
sortReplicaId(replicas);
- //always delete replica which id is minimum
- Replica deleteReplica =
tablet.getReplicaById(sortedReplicaIds.get(0));
+ List<List<Long>> newBackendsPerBucketSeq =
balance(backendsPerBucketSeq, addedBackendIds);
+ markGroupBalancing(groupId);
+ persistBackendsToBucketSeqMeta(groupId, newBackendsPerBucketSeq);
+ } catch (Exception e) {
+ LOG.error(e.getMessage(), e);
+ } finally {
+ db.readUnlock();
+ }
+ }
- long tabletSizeB = deleteReplica.getDataSize() *
partition.getMaterializedIndices().size()
- * olapTable.getPartitions().size() *
allTableIds.size();
- CloneTabletInfo tabletInfo = new
CloneTabletInfo(db.getId(), tableId, partition.getId(),
- index.getId(), tablet.getId(), (short)
replicateNum, (short) replicateNum,
- tabletSizeB, tablet.getBackendIds());
+ /**
+ * Returns a map that the key is backend id, the value is backend counter
+ * The map will sort by counter in descending order
+ *
+ * @param backends the backend id list
+ * @return a descending sorted map
+ */
+ private static Map<Long, Long> getBackendCounter(List<Long> backends) {
+ Map<Long, Long> backendCounter = backends.stream()
+ .collect(Collectors.groupingBy(Function.identity(),
Collectors.counting()));
- Long cloneReplicaBackendId =
newGroup2BackendsPerBucketSeq.get(groupId, i);
- if (cloneReplicaBackendId == null) {
- // select dest backend
- cloneReplicaBackendId = addedBackendIds.get(i
% addedBackendIds.size());
- newGroup2BackendsPerBucketSeq.put(groupId, i,
cloneReplicaBackendId);
- }
+ return
backendCounter.entrySet().stream().sorted(Collections.reverseOrder(Map.Entry.comparingByValue()))
+ .collect(Collectors.toMap(Map.Entry::getKey,
Map.Entry::getValue, (e1, e2) -> e1, LinkedHashMap::new));
+ }
- if (!colocateIndex.isGroupBalancing(groupId)) {
- colocateIndex.markGroupBalancing(groupId);
- ColocatePersistInfo info =
ColocatePersistInfo.CreateForMarkBalancing(groupId);
-
Catalog.getInstance().getEditLog().logColocateMarkBalancing(info);
- }
+ /**
+ * balance the bucket seq according to the new backends
+ *
+ * @param backendsPerBucketSeq the mapping from bucket seq to backend
+ * @param newBackends the new backends need to balance
+ * @return the balanced mapping from bucket seq to backend
+ */
+ public static List<List<Long>> balance(List<List<Long>>
backendsPerBucketSeq, List<Long> newBackends) {
+ int replicateNum = backendsPerBucketSeq.get(0).size();
+ Set<Long> groupBackendSet =
backendsPerBucketSeq.stream().flatMap(List::stream).collect(Collectors.toSet());
+ List<Long> groupBackendList =
backendsPerBucketSeq.stream().flatMap(List::stream).collect(Collectors.toList());
+ Map<Long, Long> sortedBackendCounter =
getBackendCounter(groupBackendList);
+
+ int allBackendSize = groupBackendSet.size() + newBackends.size();
+ //all backend should keep at least one replica
+ int avgReplicaNum = Math.max(groupBackendList.size() / allBackendSize,
1);
+ //the most balance case: all replica in all bucket seq have different
backend
+ int needBalanceNum = Math.min(avgReplicaNum * newBackends.size(),
groupBackendList.size() - groupBackendSet.size());
+
+ LOG.info("avg ReplicaNum: " + avgReplicaNum);
+ LOG.info("need BalanceNum: " + needBalanceNum);
+
+ int hasBalancedNum = 0;
+ //keep which BucketSeq will migrate to the new target backend
+ Map<Long, List<Integer>> targetBackendsToBucketSeqs =
Maps.newHashMap();
+
+ while (hasBalancedNum < needBalanceNum) {
+ for(Map.Entry<Long, Long> beckendCounter:
sortedBackendCounter.entrySet()) {
+ long count = beckendCounter.getValue();
+ long sourceBackend = beckendCounter.getKey();
+
+ //new backend should not as sourceBackend
+ if (newBackends.contains(sourceBackend)) {
Review comment:
has comment
----------------------------------------------------------------
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:
[email protected]
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
