You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@iotdb.apache.org by "HxpSerein (via GitHub)" <gi...@apache.org> on 2025/04/07 04:19:24 UTC
[PR] [remove datanode] GCR load balancing implement for removing datanode [iotdb]
HxpSerein opened a new pull request, #15282:
URL: https://github.com/apache/iotdb/pull/15282
## Description
Add region migration dest datanode selection algorithm for load balancing during removing datanode.
Please see https://apache-iotdb-project.feishu.cn/docx/G5aPdZJ4QozqmkxO9dsc7sEsnTe for detail design.
--
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: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [remove datanode] GCR load balancing implement for removing datanode [iotdb]
Posted by "liyuheng55555 (via GitHub)" <gi...@apache.org>.
liyuheng55555 commented on code in PR #15282:
URL: https://github.com/apache/iotdb/pull/15282#discussion_r2043842770
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/region/GreedyCopySetRegionGroupAllocator.java:
##########
Review Comment:
Does the execution of the `removeNodeReplicaSelect` function **not strongly coupled** with `GreedyCopySetRegionGroupAllocator`?
From my observation, this coupling only exists in the three class members: `regionCounter`, `databaseRegionCounter`, and `combinationCounter`.
In this case, perhaps placing `removeNodeReplicaSelect` in a new class would make things clearer, as the original class and algorithm are already complex enough.
--
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: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [remove datanode] GCR load balancing implement for removing datanode [iotdb]
Posted by "liyuheng55555 (via GitHub)" <gi...@apache.org>.
liyuheng55555 commented on code in PR #15282:
URL: https://github.com/apache/iotdb/pull/15282#discussion_r2043859227
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/region/GreedyRegionGroupAllocator.java:
##########
@@ -82,6 +83,18 @@ public TRegionReplicaSet generateOptimalRegionReplicasDistribution(
weightList.stream().limit(replicationFactor).collect(Collectors.toList()));
}
+ @Override
+ public Map<TConsensusGroupId, TDataNodeConfiguration> removeNodeReplicaSelect(
+ Map<Integer, TDataNodeConfiguration> availableDataNodeMap,
+ Map<Integer, Double> freeDiskSpaceMap,
+ List<TRegionReplicaSet> allocatedRegionGroups,
+ Map<TConsensusGroupId, String> regionDatabaseMap,
+ Map<String, List<TRegionReplicaSet>> databaseAllocatedRegionGroupMap,
+ Map<TConsensusGroupId, TRegionReplicaSet> remainReplicasMap) {
+ // TODO: Implement this method
+ return Collections.emptyMap();
Review Comment:
Throw UnsupportedOperationException will be better
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/region/PartiteGraphPlacementRegionGroupAllocator.java:
##########
@@ -114,6 +114,18 @@ public TRegionReplicaSet generateOptimalRegionReplicasDistribution(
return result;
}
+ @Override
+ public Map<TConsensusGroupId, TDataNodeConfiguration> removeNodeReplicaSelect(
+ Map<Integer, TDataNodeConfiguration> availableDataNodeMap,
+ Map<Integer, Double> freeDiskSpaceMap,
+ List<TRegionReplicaSet> allocatedRegionGroups,
+ Map<TConsensusGroupId, String> regionDatabaseMap,
+ Map<String, List<TRegionReplicaSet>> databaseAllocatedRegionGroupMap,
+ Map<TConsensusGroupId, TRegionReplicaSet> remainReplicasMap) {
+ // TODO: Implement this method
+ return Collections.emptyMap();
Review Comment:
Throw UnsupportedOperationException will be better
--
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: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [remove datanode] GCR load balancing implement for removing datanode [iotdb]
Posted by "HxpSerein (via GitHub)" <gi...@apache.org>.
HxpSerein commented on code in PR #15282:
URL: https://github.com/apache/iotdb/pull/15282#discussion_r2049813784
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/region/GreedyCopySetRegionGroupAllocator.java:
##########
@@ -125,21 +125,241 @@ public TRegionReplicaSet generateOptimalRegionReplicasDistribution(
}
}
+ @Override
+ public Map<TConsensusGroupId, TDataNodeConfiguration> removeNodeReplicaSelect(
+ Map<Integer, TDataNodeConfiguration> availableDataNodeMap,
+ Map<Integer, Double> freeDiskSpaceMap,
+ List<TRegionReplicaSet> allocatedRegionGroups,
+ Map<TConsensusGroupId, String> regionDatabaseMap,
+ Map<String, List<TRegionReplicaSet>> databaseAllocatedRegionGroupMap,
+ Map<TConsensusGroupId, TRegionReplicaSet> remainReplicasMap) {
+ try {
+ // 1. prepare: compute regionCounter, databaseRegionCounter, and combinationCounter
+
+ List<TRegionReplicaSet> databaseAllocatedRegionGroups =
+ new ArrayList<>(databaseAllocatedRegionGroupMap.values()).get(0);
+ prepare(availableDataNodeMap, allocatedRegionGroups, databaseAllocatedRegionGroups);
+
+ // 2. Build allowed candidate set for each region that needs to be migrated.
+ // For each region in remainReplicasMap, the candidate destination nodes are all nodes in
+ // availableDataNodeMap
+ // excluding those already in the remain replica set.
+ List<TConsensusGroupId> regionKeys = new ArrayList<>(remainReplicasMap.keySet());
+ Map<TConsensusGroupId, List<Integer>> allowedCandidatesMap = new HashMap<>();
+ for (TConsensusGroupId regionId : regionKeys) {
+ TRegionReplicaSet remainReplicaSet = remainReplicasMap.get(regionId);
+ Set<Integer> notAllowedNodes = new HashSet<>();
+
+ // Exclude nodes already present in the remain replica set
+ for (TDataNodeLocation location : remainReplicaSet.getDataNodeLocations()) {
+ notAllowedNodes.add(location.getDataNodeId());
+ }
+
+ // Allowed candidates are the nodes not in the exclusion set
+ List<Integer> candidates =
+ availableDataNodeMap.keySet().stream()
+ .filter(nodeId -> !notAllowedNodes.contains(nodeId))
+ .sorted(
+ (a, b) -> {
+ int cmp = Integer.compare(regionCounter[a], regionCounter[b]);
+ if (cmp == 0) {
+ cmp = Integer.compare(databaseRegionCounter[a], databaseRegionCounter[b]);
+ }
+ return cmp;
+ })
+ .collect(Collectors.toList());
+
+ // Sort candidates in ascending order of current global load (regionCounter)
+ allowedCandidatesMap.put(regionId, candidates);
+ }
+
+ // Optionally, sort regionKeys by the size of its candidate list (smaller candidate sets
+ // first)
+ regionKeys.sort(Comparator.comparingInt(id -> allowedCandidatesMap.get(id).size()));
+
+ int n = regionKeys.size();
+ // Each element holds the candidate nodeId chosen for the region at that index
+ int[] currentAssignment = new int[n];
+ // additionalLoad holds the number of extra regions assigned to each node in this migration
+ // solution.
+ int[] additionalLoad = new int[regionCounter.length];
+
+ // 3. Create a buffer for candidate solutions
+ List<int[]> optimalAssignments = new ArrayList<>();
+ // bestMetrics holds the best found metrics: [maxGlobalLoad, maxDatabaseLoad, scatterValue].
+ // Initialize to high values.
+ int[] bestMetrics = new int[] {Integer.MAX_VALUE, Integer.MAX_VALUE, Integer.MAX_VALUE};
+
+ dfsRemoveNodeReplica(
+ 0,
+ regionKeys,
+ allowedCandidatesMap,
+ currentAssignment,
+ additionalLoad,
+ optimalAssignments,
+ bestMetrics,
+ remainReplicasMap);
+
+ // 4. Randomly select one solution from the candidate buffer
+ if (optimalAssignments.isEmpty()) {
+ // This should not happen if there is at least one valid assignment
+ return Collections.emptyMap();
+ }
+ Collections.shuffle(optimalAssignments);
+ int[] bestAssignment = optimalAssignments.get(0);
+
+ // 5. Build and return the result mapping: region -> chosen destination TDataNodeConfiguration
+ Map<TConsensusGroupId, TDataNodeConfiguration> result = new HashMap<>();
+ for (int i = 0; i < n; i++) {
+ TConsensusGroupId regionId = regionKeys.get(i);
+ int chosenNodeId = bestAssignment[i];
+ result.put(regionId, availableDataNodeMap.get(chosenNodeId));
+ }
+ return result;
+ } finally {
+ // Clear any temporary state to avoid impacting subsequent calls
+ clear();
+ }
+ }
+
+ /**
+ * DFS method that searches for migration target assignments.
+ *
+ * <p>It enumerates all possible assignments (one candidate for each region) and collects
+ * candidate solutions in the optimalAssignments buffer. The evaluation metrics for each complete
+ * assignment (i.e. when index == regionKeys.size()) are:
+ *
+ * <p>1. Max global load: the maximum over nodes of (regionCounter[node] + additionalLoad[node])
+ * 2. Max database load: the maximum over nodes of (databaseRegionCounter[node] +
+ * additionalLoad[node]) 3. Scatter value: computed per region, summing the combinationCounter for
+ * every pair in the complete replica set. The complete replica set for a region includes nodes in
+ * its remain replica set plus the newly assigned node.
+ *
+ * <p>The candidates are compared lexicographically (first by global load, then by database load,
+ * then by scatter). When a better candidate is found, the optimalAssignments buffer is cleared
+ * and updated; if the new candidate matches the best found metrics, it is added to the buffer.
+ *
+ * <p>DFS search is pruned if the optimalAssignments buffer reaches CAPACITY.
+ *
+ * @param index Current DFS level, corresponding to regionKeys.get(index)
+ * @param regionKeys A list of regions that need to be migrated.
+ * @param allowedCandidatesMap For each region, the allowed candidate destination node IDs.
+ * @param currentAssignment Current partial assignment; its length equals the number of regions.
+ * @param additionalLoad Extra load currently assigned to each node.
+ * @param optimalAssignments Buffer holding candidate assignment arrays.
+ * @param bestMetrics An int array holding the best metrics found so far: [maxGlobalLoad,
+ * maxDatabaseLoad, scatterValue].
+ * @param remainReplicasMap Mapping from region to its current remain replica set.
+ */
+ private void dfsRemoveNodeReplica(
+ int index,
+ List<TConsensusGroupId> regionKeys,
+ Map<TConsensusGroupId, List<Integer>> allowedCandidatesMap,
+ int[] currentAssignment,
+ int[] additionalLoad,
+ List<int[]> optimalAssignments,
+ int[] bestMetrics,
+ Map<TConsensusGroupId, TRegionReplicaSet> remainReplicasMap) {
+ int n = regionKeys.size();
+ if (index == n) {
+ // A complete assignment has been generated.
+ // Compute metrics for this complete migration assignment.
+
+ // Compute the scatter value for the complete assignment.
+ int currentScatter = 0;
+ // For each region, calculate the scatter based on the combinationCounter among all nodes
+ // in the full replica set (which includes the nodes in the remain replica and the new
+ // candidate).
+ for (int r = 0; r < n; r++) {
+ TConsensusGroupId regionId = regionKeys.get(r);
+ for (TDataNodeLocation location : remainReplicasMap.get(regionId).getDataNodeLocations()) {
+ int nodeA = currentAssignment[r];
+ int nodeB = location.getDataNodeId();
+ currentScatter += combinationCounter[nodeA][nodeB];
+ }
+ }
+
+ // Compute the maximum global load and maximum database load among all nodes that received
+ // additional load.
+ int[] currentMetrics = getCurrentMetrics(additionalLoad, currentScatter);
+
+ // Lexicographically compare currentMetrics with bestMetrics.
+ // If currentMetrics is better than bestMetrics, update bestMetrics and clear the candidate
+ // buffer.
+ boolean isBetter = false;
+ boolean isEqual = true;
+ for (int i = 0; i < 3; i++) {
+ if (currentMetrics[i] < bestMetrics[i]) {
+ isBetter = true;
+ isEqual = false;
+ break;
+ } else if (currentMetrics[i] > bestMetrics[i]) {
+ isEqual = false;
+ break;
+ }
+ }
+ if (isBetter) {
+ bestMetrics[0] = currentMetrics[0];
+ bestMetrics[1] = currentMetrics[1];
+ bestMetrics[2] = currentMetrics[2];
+ optimalAssignments.clear();
+ optimalAssignments.add(Arrays.copyOf(currentAssignment, n));
+ } else if (isEqual) {
+ optimalAssignments.add(Arrays.copyOf(currentAssignment, n));
+ // Prune search if we already have enough candidate solutions
+ if (optimalAssignments.size() >= GCR_MAX_OPTIMAL_PLAN_NUM) {
Review Comment:
Yes, you are right. DFS will try its best to find the best solution. It is not guaranteed to find it, but in most cases, it works well.
--
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: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [remove datanode] GCR load balancing implement for removing datanode [iotdb]
Posted by "liyuheng55555 (via GitHub)" <gi...@apache.org>.
liyuheng55555 commented on code in PR #15282:
URL: https://github.com/apache/iotdb/pull/15282#discussion_r2043842770
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/region/GreedyCopySetRegionGroupAllocator.java:
##########
Review Comment:
Does the execution of the `removeNodeReplicaSelect` function **not strongly coupled** with `GreedyCopySetRegionGroupAllocator`?
From my observation, this coupling only exists in the three class members: `regionCounter`, `databaseRegionCounter`, and `combinationCounter`.
In this case, perhaps placing `removeNodeReplicaSelect` in a another class would make things clearer, as the original class and algorithm are already complex enough.
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/region/GreedyCopySetRegionGroupAllocator.java:
##########
@@ -125,21 +125,241 @@ public TRegionReplicaSet generateOptimalRegionReplicasDistribution(
}
}
+ @Override
+ public Map<TConsensusGroupId, TDataNodeConfiguration> removeNodeReplicaSelect(
+ Map<Integer, TDataNodeConfiguration> availableDataNodeMap,
+ Map<Integer, Double> freeDiskSpaceMap,
+ List<TRegionReplicaSet> allocatedRegionGroups,
+ Map<TConsensusGroupId, String> regionDatabaseMap,
+ Map<String, List<TRegionReplicaSet>> databaseAllocatedRegionGroupMap,
+ Map<TConsensusGroupId, TRegionReplicaSet> remainReplicasMap) {
+ try {
+ // 1. prepare: compute regionCounter, databaseRegionCounter, and combinationCounter
+
+ List<TRegionReplicaSet> databaseAllocatedRegionGroups =
+ new ArrayList<>(databaseAllocatedRegionGroupMap.values()).get(0);
+ prepare(availableDataNodeMap, allocatedRegionGroups, databaseAllocatedRegionGroups);
+
+ // 2. Build allowed candidate set for each region that needs to be migrated.
+ // For each region in remainReplicasMap, the candidate destination nodes are all nodes in
+ // availableDataNodeMap
+ // excluding those already in the remain replica set.
+ List<TConsensusGroupId> regionKeys = new ArrayList<>(remainReplicasMap.keySet());
+ Map<TConsensusGroupId, List<Integer>> allowedCandidatesMap = new HashMap<>();
+ for (TConsensusGroupId regionId : regionKeys) {
+ TRegionReplicaSet remainReplicaSet = remainReplicasMap.get(regionId);
+ Set<Integer> notAllowedNodes = new HashSet<>();
+
+ // Exclude nodes already present in the remain replica set
+ for (TDataNodeLocation location : remainReplicaSet.getDataNodeLocations()) {
+ notAllowedNodes.add(location.getDataNodeId());
+ }
+
+ // Allowed candidates are the nodes not in the exclusion set
+ List<Integer> candidates =
+ availableDataNodeMap.keySet().stream()
+ .filter(nodeId -> !notAllowedNodes.contains(nodeId))
+ .sorted(
+ (a, b) -> {
+ int cmp = Integer.compare(regionCounter[a], regionCounter[b]);
+ if (cmp == 0) {
+ cmp = Integer.compare(databaseRegionCounter[a], databaseRegionCounter[b]);
+ }
+ return cmp;
+ })
+ .collect(Collectors.toList());
+
+ // Sort candidates in ascending order of current global load (regionCounter)
+ allowedCandidatesMap.put(regionId, candidates);
+ }
+
+ // Optionally, sort regionKeys by the size of its candidate list (smaller candidate sets
+ // first)
+ regionKeys.sort(Comparator.comparingInt(id -> allowedCandidatesMap.get(id).size()));
+
+ int n = regionKeys.size();
+ // Each element holds the candidate nodeId chosen for the region at that index
+ int[] currentAssignment = new int[n];
+ // additionalLoad holds the number of extra regions assigned to each node in this migration
+ // solution.
+ int[] additionalLoad = new int[regionCounter.length];
+
+ // 3. Create a buffer for candidate solutions
+ List<int[]> optimalAssignments = new ArrayList<>();
+ // bestMetrics holds the best found metrics: [maxGlobalLoad, maxDatabaseLoad, scatterValue].
+ // Initialize to high values.
+ int[] bestMetrics = new int[] {Integer.MAX_VALUE, Integer.MAX_VALUE, Integer.MAX_VALUE};
+
+ dfsRemoveNodeReplica(
+ 0,
+ regionKeys,
+ allowedCandidatesMap,
+ currentAssignment,
+ additionalLoad,
+ optimalAssignments,
+ bestMetrics,
+ remainReplicasMap);
+
+ // 4. Randomly select one solution from the candidate buffer
+ if (optimalAssignments.isEmpty()) {
+ // This should not happen if there is at least one valid assignment
+ return Collections.emptyMap();
+ }
+ Collections.shuffle(optimalAssignments);
+ int[] bestAssignment = optimalAssignments.get(0);
+
+ // 5. Build and return the result mapping: region -> chosen destination TDataNodeConfiguration
+ Map<TConsensusGroupId, TDataNodeConfiguration> result = new HashMap<>();
+ for (int i = 0; i < n; i++) {
+ TConsensusGroupId regionId = regionKeys.get(i);
+ int chosenNodeId = bestAssignment[i];
+ result.put(regionId, availableDataNodeMap.get(chosenNodeId));
+ }
+ return result;
+ } finally {
+ // Clear any temporary state to avoid impacting subsequent calls
+ clear();
+ }
+ }
+
+ /**
+ * DFS method that searches for migration target assignments.
+ *
+ * <p>It enumerates all possible assignments (one candidate for each region) and collects
+ * candidate solutions in the optimalAssignments buffer. The evaluation metrics for each complete
+ * assignment (i.e. when index == regionKeys.size()) are:
+ *
+ * <p>1. Max global load: the maximum over nodes of (regionCounter[node] + additionalLoad[node])
+ * 2. Max database load: the maximum over nodes of (databaseRegionCounter[node] +
+ * additionalLoad[node]) 3. Scatter value: computed per region, summing the combinationCounter for
+ * every pair in the complete replica set. The complete replica set for a region includes nodes in
+ * its remain replica set plus the newly assigned node.
+ *
+ * <p>The candidates are compared lexicographically (first by global load, then by database load,
+ * then by scatter). When a better candidate is found, the optimalAssignments buffer is cleared
+ * and updated; if the new candidate matches the best found metrics, it is added to the buffer.
+ *
+ * <p>DFS search is pruned if the optimalAssignments buffer reaches CAPACITY.
+ *
+ * @param index Current DFS level, corresponding to regionKeys.get(index)
+ * @param regionKeys A list of regions that need to be migrated.
+ * @param allowedCandidatesMap For each region, the allowed candidate destination node IDs.
+ * @param currentAssignment Current partial assignment; its length equals the number of regions.
+ * @param additionalLoad Extra load currently assigned to each node.
+ * @param optimalAssignments Buffer holding candidate assignment arrays.
+ * @param bestMetrics An int array holding the best metrics found so far: [maxGlobalLoad,
+ * maxDatabaseLoad, scatterValue].
+ * @param remainReplicasMap Mapping from region to its current remain replica set.
+ */
+ private void dfsRemoveNodeReplica(
+ int index,
+ List<TConsensusGroupId> regionKeys,
+ Map<TConsensusGroupId, List<Integer>> allowedCandidatesMap,
+ int[] currentAssignment,
+ int[] additionalLoad,
+ List<int[]> optimalAssignments,
+ int[] bestMetrics,
+ Map<TConsensusGroupId, TRegionReplicaSet> remainReplicasMap) {
+ int n = regionKeys.size();
+ if (index == n) {
+ // A complete assignment has been generated.
+ // Compute metrics for this complete migration assignment.
+
+ // Compute the scatter value for the complete assignment.
+ int currentScatter = 0;
+ // For each region, calculate the scatter based on the combinationCounter among all nodes
+ // in the full replica set (which includes the nodes in the remain replica and the new
+ // candidate).
+ for (int r = 0; r < n; r++) {
+ TConsensusGroupId regionId = regionKeys.get(r);
+ for (TDataNodeLocation location : remainReplicasMap.get(regionId).getDataNodeLocations()) {
+ int nodeA = currentAssignment[r];
+ int nodeB = location.getDataNodeId();
+ currentScatter += combinationCounter[nodeA][nodeB];
+ }
+ }
+
+ // Compute the maximum global load and maximum database load among all nodes that received
+ // additional load.
+ int[] currentMetrics = getCurrentMetrics(additionalLoad, currentScatter);
+
+ // Lexicographically compare currentMetrics with bestMetrics.
+ // If currentMetrics is better than bestMetrics, update bestMetrics and clear the candidate
+ // buffer.
+ boolean isBetter = false;
+ boolean isEqual = true;
+ for (int i = 0; i < 3; i++) {
+ if (currentMetrics[i] < bestMetrics[i]) {
+ isBetter = true;
+ isEqual = false;
+ break;
+ } else if (currentMetrics[i] > bestMetrics[i]) {
+ isEqual = false;
+ break;
+ }
+ }
+ if (isBetter) {
+ bestMetrics[0] = currentMetrics[0];
+ bestMetrics[1] = currentMetrics[1];
+ bestMetrics[2] = currentMetrics[2];
+ optimalAssignments.clear();
+ optimalAssignments.add(Arrays.copyOf(currentAssignment, n));
+ } else if (isEqual) {
+ optimalAssignments.add(Arrays.copyOf(currentAssignment, n));
+ // Prune search if we already have enough candidate solutions
+ if (optimalAssignments.size() >= GCR_MAX_OPTIMAL_PLAN_NUM) {
Review Comment:
It seems that the DFS will end when the number of search results reaches `GCR_MAX_OPTIMAL_PLAN_NUM`.
Does the algorithm always guarantee finding the optimal solutions? If the algorithm first finds `GCR_MAX_OPTIMAL_PLAN_NUM` suboptimal solutions, will the DFS end?
--
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: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [remove datanode] GCR load balancing implement for removing datanode [iotdb]
Posted by "HxpSerein (via GitHub)" <gi...@apache.org>.
HxpSerein commented on code in PR #15282:
URL: https://github.com/apache/iotdb/pull/15282#discussion_r2049813784
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/region/GreedyCopySetRegionGroupAllocator.java:
##########
@@ -125,21 +125,241 @@ public TRegionReplicaSet generateOptimalRegionReplicasDistribution(
}
}
+ @Override
+ public Map<TConsensusGroupId, TDataNodeConfiguration> removeNodeReplicaSelect(
+ Map<Integer, TDataNodeConfiguration> availableDataNodeMap,
+ Map<Integer, Double> freeDiskSpaceMap,
+ List<TRegionReplicaSet> allocatedRegionGroups,
+ Map<TConsensusGroupId, String> regionDatabaseMap,
+ Map<String, List<TRegionReplicaSet>> databaseAllocatedRegionGroupMap,
+ Map<TConsensusGroupId, TRegionReplicaSet> remainReplicasMap) {
+ try {
+ // 1. prepare: compute regionCounter, databaseRegionCounter, and combinationCounter
+
+ List<TRegionReplicaSet> databaseAllocatedRegionGroups =
+ new ArrayList<>(databaseAllocatedRegionGroupMap.values()).get(0);
+ prepare(availableDataNodeMap, allocatedRegionGroups, databaseAllocatedRegionGroups);
+
+ // 2. Build allowed candidate set for each region that needs to be migrated.
+ // For each region in remainReplicasMap, the candidate destination nodes are all nodes in
+ // availableDataNodeMap
+ // excluding those already in the remain replica set.
+ List<TConsensusGroupId> regionKeys = new ArrayList<>(remainReplicasMap.keySet());
+ Map<TConsensusGroupId, List<Integer>> allowedCandidatesMap = new HashMap<>();
+ for (TConsensusGroupId regionId : regionKeys) {
+ TRegionReplicaSet remainReplicaSet = remainReplicasMap.get(regionId);
+ Set<Integer> notAllowedNodes = new HashSet<>();
+
+ // Exclude nodes already present in the remain replica set
+ for (TDataNodeLocation location : remainReplicaSet.getDataNodeLocations()) {
+ notAllowedNodes.add(location.getDataNodeId());
+ }
+
+ // Allowed candidates are the nodes not in the exclusion set
+ List<Integer> candidates =
+ availableDataNodeMap.keySet().stream()
+ .filter(nodeId -> !notAllowedNodes.contains(nodeId))
+ .sorted(
+ (a, b) -> {
+ int cmp = Integer.compare(regionCounter[a], regionCounter[b]);
+ if (cmp == 0) {
+ cmp = Integer.compare(databaseRegionCounter[a], databaseRegionCounter[b]);
+ }
+ return cmp;
+ })
+ .collect(Collectors.toList());
+
+ // Sort candidates in ascending order of current global load (regionCounter)
+ allowedCandidatesMap.put(regionId, candidates);
+ }
+
+ // Optionally, sort regionKeys by the size of its candidate list (smaller candidate sets
+ // first)
+ regionKeys.sort(Comparator.comparingInt(id -> allowedCandidatesMap.get(id).size()));
+
+ int n = regionKeys.size();
+ // Each element holds the candidate nodeId chosen for the region at that index
+ int[] currentAssignment = new int[n];
+ // additionalLoad holds the number of extra regions assigned to each node in this migration
+ // solution.
+ int[] additionalLoad = new int[regionCounter.length];
+
+ // 3. Create a buffer for candidate solutions
+ List<int[]> optimalAssignments = new ArrayList<>();
+ // bestMetrics holds the best found metrics: [maxGlobalLoad, maxDatabaseLoad, scatterValue].
+ // Initialize to high values.
+ int[] bestMetrics = new int[] {Integer.MAX_VALUE, Integer.MAX_VALUE, Integer.MAX_VALUE};
+
+ dfsRemoveNodeReplica(
+ 0,
+ regionKeys,
+ allowedCandidatesMap,
+ currentAssignment,
+ additionalLoad,
+ optimalAssignments,
+ bestMetrics,
+ remainReplicasMap);
+
+ // 4. Randomly select one solution from the candidate buffer
+ if (optimalAssignments.isEmpty()) {
+ // This should not happen if there is at least one valid assignment
+ return Collections.emptyMap();
+ }
+ Collections.shuffle(optimalAssignments);
+ int[] bestAssignment = optimalAssignments.get(0);
+
+ // 5. Build and return the result mapping: region -> chosen destination TDataNodeConfiguration
+ Map<TConsensusGroupId, TDataNodeConfiguration> result = new HashMap<>();
+ for (int i = 0; i < n; i++) {
+ TConsensusGroupId regionId = regionKeys.get(i);
+ int chosenNodeId = bestAssignment[i];
+ result.put(regionId, availableDataNodeMap.get(chosenNodeId));
+ }
+ return result;
+ } finally {
+ // Clear any temporary state to avoid impacting subsequent calls
+ clear();
+ }
+ }
+
+ /**
+ * DFS method that searches for migration target assignments.
+ *
+ * <p>It enumerates all possible assignments (one candidate for each region) and collects
+ * candidate solutions in the optimalAssignments buffer. The evaluation metrics for each complete
+ * assignment (i.e. when index == regionKeys.size()) are:
+ *
+ * <p>1. Max global load: the maximum over nodes of (regionCounter[node] + additionalLoad[node])
+ * 2. Max database load: the maximum over nodes of (databaseRegionCounter[node] +
+ * additionalLoad[node]) 3. Scatter value: computed per region, summing the combinationCounter for
+ * every pair in the complete replica set. The complete replica set for a region includes nodes in
+ * its remain replica set plus the newly assigned node.
+ *
+ * <p>The candidates are compared lexicographically (first by global load, then by database load,
+ * then by scatter). When a better candidate is found, the optimalAssignments buffer is cleared
+ * and updated; if the new candidate matches the best found metrics, it is added to the buffer.
+ *
+ * <p>DFS search is pruned if the optimalAssignments buffer reaches CAPACITY.
+ *
+ * @param index Current DFS level, corresponding to regionKeys.get(index)
+ * @param regionKeys A list of regions that need to be migrated.
+ * @param allowedCandidatesMap For each region, the allowed candidate destination node IDs.
+ * @param currentAssignment Current partial assignment; its length equals the number of regions.
+ * @param additionalLoad Extra load currently assigned to each node.
+ * @param optimalAssignments Buffer holding candidate assignment arrays.
+ * @param bestMetrics An int array holding the best metrics found so far: [maxGlobalLoad,
+ * maxDatabaseLoad, scatterValue].
+ * @param remainReplicasMap Mapping from region to its current remain replica set.
+ */
+ private void dfsRemoveNodeReplica(
+ int index,
+ List<TConsensusGroupId> regionKeys,
+ Map<TConsensusGroupId, List<Integer>> allowedCandidatesMap,
+ int[] currentAssignment,
+ int[] additionalLoad,
+ List<int[]> optimalAssignments,
+ int[] bestMetrics,
+ Map<TConsensusGroupId, TRegionReplicaSet> remainReplicasMap) {
+ int n = regionKeys.size();
+ if (index == n) {
+ // A complete assignment has been generated.
+ // Compute metrics for this complete migration assignment.
+
+ // Compute the scatter value for the complete assignment.
+ int currentScatter = 0;
+ // For each region, calculate the scatter based on the combinationCounter among all nodes
+ // in the full replica set (which includes the nodes in the remain replica and the new
+ // candidate).
+ for (int r = 0; r < n; r++) {
+ TConsensusGroupId regionId = regionKeys.get(r);
+ for (TDataNodeLocation location : remainReplicasMap.get(regionId).getDataNodeLocations()) {
+ int nodeA = currentAssignment[r];
+ int nodeB = location.getDataNodeId();
+ currentScatter += combinationCounter[nodeA][nodeB];
+ }
+ }
+
+ // Compute the maximum global load and maximum database load among all nodes that received
+ // additional load.
+ int[] currentMetrics = getCurrentMetrics(additionalLoad, currentScatter);
+
+ // Lexicographically compare currentMetrics with bestMetrics.
+ // If currentMetrics is better than bestMetrics, update bestMetrics and clear the candidate
+ // buffer.
+ boolean isBetter = false;
+ boolean isEqual = true;
+ for (int i = 0; i < 3; i++) {
+ if (currentMetrics[i] < bestMetrics[i]) {
+ isBetter = true;
+ isEqual = false;
+ break;
+ } else if (currentMetrics[i] > bestMetrics[i]) {
+ isEqual = false;
+ break;
+ }
+ }
+ if (isBetter) {
+ bestMetrics[0] = currentMetrics[0];
+ bestMetrics[1] = currentMetrics[1];
+ bestMetrics[2] = currentMetrics[2];
+ optimalAssignments.clear();
+ optimalAssignments.add(Arrays.copyOf(currentAssignment, n));
+ } else if (isEqual) {
+ optimalAssignments.add(Arrays.copyOf(currentAssignment, n));
+ // Prune search if we already have enough candidate solutions
+ if (optimalAssignments.size() >= GCR_MAX_OPTIMAL_PLAN_NUM) {
Review Comment:
Yes, you are right. DFS will try its best to find the best solution, but it cannot guarantee that it will find the best solution.
--
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: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [remove datanode] GCR load balancing implement for removing datanode [iotdb]
Posted by "CRZbulabula (via GitHub)" <gi...@apache.org>.
CRZbulabula merged PR #15282:
URL: https://github.com/apache/iotdb/pull/15282
--
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: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org