imay 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_r282854896
##########
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));
Review comment:
Hash map is not ordered, I think you what you need is List<Map.Entry<Long,
Long>>
----------------------------------------------------------------
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]
