You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@iotdb.apache.org by "CRZbulabula (via GitHub)" <gi...@apache.org> on 2023/07/21 12:42:40 UTC
[GitHub] [iotdb] CRZbulabula opened a new pull request, #10648: Computing resource balancing
CRZbulabula opened a new pull request, #10648:
URL: https://github.com/apache/iotdb/pull/10648
Draft PR for IT
--
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
[GitHub] [iotdb] CRZbulabula commented on a diff in pull request #10648: [IOTDB-6079] Cluster computing resource balance
Posted by "CRZbulabula (via GitHub)" <gi...@apache.org>.
CRZbulabula commented on code in PR #10648:
URL: https://github.com/apache/iotdb/pull/10648#discussion_r1278955319
##########
iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/partition/DataPartitionEntry.java:
##########
@@ -0,0 +1,87 @@
+/*
+ * 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.iotdb.commons.partition;
+
+import org.apache.iotdb.common.rpc.thrift.TConsensusGroupId;
+import org.apache.iotdb.common.rpc.thrift.TSeriesPartitionSlot;
+import org.apache.iotdb.common.rpc.thrift.TTimePartitionSlot;
+
+import java.security.SecureRandom;
+import java.util.Objects;
+
+public class DataPartitionEntry implements Comparable<DataPartitionEntry> {
Review Comment:
Deleted now
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/statemachine/CreateRegionGroupsProcedure.java:
##########
@@ -259,6 +275,7 @@ public void deserialize(ByteBuffer byteBuffer) {
ThriftCommonsSerDeUtils.deserializeTRegionReplicaSet(byteBuffer);
failedRegionReplicaSets.put(groupId, replica);
}
+ persistPlan.deserializeForProcedure(byteBuffer);
Review Comment:
We can ignore the compatibility requirement of Procedure. Because there must no serialized Procedures when doiong upgration
--
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
[GitHub] [iotdb] Cpaulyz commented on a diff in pull request #10648: [IOTDB-6079] Cluster computing resource balance
Posted by "Cpaulyz (via GitHub)" <gi...@apache.org>.
Cpaulyz commented on code in PR #10648:
URL: https://github.com/apache/iotdb/pull/10648#discussion_r1278922379
##########
integration-test/src/test/java/org/apache/iotdb/session/it/IoTDBSessionSchemaTemplateIT.java:
##########
@@ -471,18 +471,22 @@ public void testHybridAutoExtendSchemaTemplate()
Assert.assertTrue(expectedSeries.isEmpty());
- deviceIds = Arrays.asList("root.db.v4.d1", "root.db.v4.d2");
- timestamps = Arrays.asList(1L, 1L);
measurements = Arrays.asList("a", "b", "c");
- allMeasurements = Arrays.asList(measurements, measurements);
- allTsDataTypes =
- Arrays.asList(
- Arrays.asList(TSDataType.FLOAT, TSDataType.FLOAT, TSDataType.TEXT),
- Arrays.asList(TSDataType.DOUBLE, TSDataType.DOUBLE, TSDataType.INT32));
- allValues = Arrays.asList(Arrays.asList(1f, 2f, "3"), Arrays.asList(1d, 2d, 3));
try {
- session.insertRecords(deviceIds, timestamps, allMeasurements, allTsDataTypes, allValues);
+ session.insertRecord(
Review Comment:
The new partitioning strategy puts the two devices in two partitions instead of one. insertRecords triggers the write redirection, and the order of the two writes operation is randomized.
--
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
[GitHub] [iotdb] OneSizeFitsQuorum commented on a diff in pull request #10648: [IOTDB-6079] Cluster computing resource balance
Posted by "OneSizeFitsQuorum (via GitHub)" <gi...@apache.org>.
OneSizeFitsQuorum commented on code in PR #10648:
URL: https://github.com/apache/iotdb/pull/10648#discussion_r1278981876
##########
integration-test/src/test/java/org/apache/iotdb/confignode/it/partition/IoTDBPartitionCreationIT.java:
##########
@@ -419,182 +419,6 @@ public void testPartitionAllocation() throws Exception {
4 + testTimePartitionBatchSize,
testTimePartitionInterval,
dataPartitionTableResp.getDataPartitionTable());
-
Review Comment:
OK
--
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
[GitHub] [iotdb] OneSizeFitsQuorum merged pull request #10648: [IOTDB-6079] Cluster computing resource balance
Posted by "OneSizeFitsQuorum (via GitHub)" <gi...@apache.org>.
OneSizeFitsQuorum merged PR #10648:
URL: https://github.com/apache/iotdb/pull/10648
--
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
[GitHub] [iotdb] sonarcloud[bot] commented on pull request #10648: [IOTDB-6079] Cluster computing resource balance
Posted by "sonarcloud[bot] (via GitHub)" <gi...@apache.org>.
sonarcloud[bot] commented on PR #10648:
URL: https://github.com/apache/iotdb/pull/10648#issuecomment-1657921585
SonarCloud Quality Gate failed. [](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=10648)
[](https://sonarcloud.io/project/issues?id=apache_iotdb&pullRequest=10648&resolved=false&types=BUG) [](https://sonarcloud.io/project/issues?id=apache_iotdb&pullRequest=10648&resolved=false&types=BUG) [0 Bugs](https://sonarcloud.io/project/issues?id=apache_iotdb&pullRequest=10648&resolved=false&types=BUG)
[](https://sonarcloud.io/project/issues?id=apache_iotdb&pullRequest=10648&resolved=false&types=VULNERABILITY) [](https://sonarcloud.io/project/issues?id=apache_iotdb&pullRequest=10648&resolved=false&types=VULNERABILITY) [0 Vulnerabilities](https://sonarcloud.io/project/issues?id=apache_iotdb&pullRequest=10648&resolved=false&types=VULNERABILITY)
[](https://sonarcloud.io/project/security_hotspots?id=apache_iotdb&pullRequest=10648&resolved=false&types=SECURITY_HOTSPOT) [](https://sonarcloud.io/project/security_hotspots?id=apache_iotdb&pullRequest=10648&resolved=false&types=SECURITY_HOTSPOT) [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=apache_iotdb&pullRequest=10648&resolved=false&types=SECURITY_HOTSPOT)
[](https://sonarcloud.io/project/issues?id=apache_iotdb&pullRequest=10648&resolved=false&types=CODE_SMELL) [](https://sonarcloud.io/project/issues?id=apache_iotdb&pullRequest=10648&resolved=false&types=CODE_SMELL) [7 Code Smells](https://sonarcloud.io/project/issues?id=apache_iotdb&pullRequest=10648&resolved=false&types=CODE_SMELL)
[](https://sonarcloud.io/component_measures?id=apache_iotdb&pullRequest=10648&metric=new_coverage&view=list) [0.0% Coverage](https://sonarcloud.io/component_measures?id=apache_iotdb&pullRequest=10648&metric=new_coverage&view=list)
[](https://sonarcloud.io/component_measures?id=apache_iotdb&pullRequest=10648&metric=new_duplicated_lines_density&view=list) [0.0% Duplication](https://sonarcloud.io/component_measures?id=apache_iotdb&pullRequest=10648&metric=new_duplicated_lines_density&view=list)
 Catch issues before they fail your Quality Gate with our IDE extension  [SonarLint](https://www.sonarsource.com/products/sonarlint/features/connected-mode/?referrer=sonarcloud-welcome)
--
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
[GitHub] [iotdb] OneSizeFitsQuorum commented on a diff in pull request #10648: [IOTDB-6079] Cluster computing resource balance
Posted by "OneSizeFitsQuorum (via GitHub)" <gi...@apache.org>.
OneSizeFitsQuorum commented on code in PR #10648:
URL: https://github.com/apache/iotdb/pull/10648#discussion_r1278773349
##########
integration-test/src/test/java/org/apache/iotdb/session/it/IoTDBSessionSchemaTemplateIT.java:
##########
@@ -471,18 +471,22 @@ public void testHybridAutoExtendSchemaTemplate()
Assert.assertTrue(expectedSeries.isEmpty());
- deviceIds = Arrays.asList("root.db.v4.d1", "root.db.v4.d2");
- timestamps = Arrays.asList(1L, 1L);
measurements = Arrays.asList("a", "b", "c");
- allMeasurements = Arrays.asList(measurements, measurements);
- allTsDataTypes =
- Arrays.asList(
- Arrays.asList(TSDataType.FLOAT, TSDataType.FLOAT, TSDataType.TEXT),
- Arrays.asList(TSDataType.DOUBLE, TSDataType.DOUBLE, TSDataType.INT32));
- allValues = Arrays.asList(Arrays.asList(1f, 2f, "3"), Arrays.asList(1d, 2d, 3));
try {
- session.insertRecords(deviceIds, timestamps, allMeasurements, allTsDataTypes, allValues);
+ session.insertRecord(
Review Comment:
I'm kind of curious why we change here? Their logic seems to be the same
##########
integration-test/src/test/java/org/apache/iotdb/confignode/it/partition/IoTDBPartitionInheritPolicyIT.java:
##########
@@ -95,76 +91,173 @@ public static void tearDown() {
@Test
public void testDataPartitionInheritPolicy() throws Exception {
+ final long baseStartTime = 1000;
+ Map<TSeriesPartitionSlot, TConsensusGroupId> dataAllotTable1 = new ConcurrentHashMap<>();
- try (SyncConfigNodeIServiceClient client =
- (SyncConfigNodeIServiceClient) EnvFactory.getEnv().getLeaderConfigNodeConnection()) {
- TDataPartitionReq dataPartitionReq = new TDataPartitionReq();
- TDataPartitionTableResp dataPartitionTableResp;
- Map<String, Map<TSeriesPartitionSlot, TTimeSlotList>> partitionSlotsMap;
-
- for (int i = 0; i < storageGroupNum; i++) {
- String storageGroup = sg + i;
- for (int j = 0; j < testSeriesPartitionSlotNum; j += seriesPartitionBatchSize) {
- // Test inherit predecessor or successor
- boolean isAscending = (j / 10) % 2 == 0;
- int step = isAscending ? timePartitionBatchSize : -timePartitionBatchSize;
- int k = isAscending ? 0 : testTimePartitionSlotsNum - timePartitionBatchSize;
- while (0 <= k && k < testTimePartitionSlotsNum) {
- partitionSlotsMap =
- ConfigNodeTestUtils.constructPartitionSlotsMap(
- storageGroup,
- j,
- j + seriesPartitionBatchSize,
- k,
- k + timePartitionBatchSize,
- testTimePartitionInterval);
- // Let ConfigNode create DataPartition
- dataPartitionReq.setPartitionSlotsMap(partitionSlotsMap);
- for (int retry = 0; retry < 5; retry++) {
- // Build new Client since it's unstable
- try (SyncConfigNodeIServiceClient configNodeClient =
- (SyncConfigNodeIServiceClient)
- EnvFactory.getEnv().getLeaderConfigNodeConnection()) {
- dataPartitionTableResp =
- configNodeClient.getOrCreateDataPartitionTable(dataPartitionReq);
- if (dataPartitionTableResp != null
- && dataPartitionTableResp.getStatus().getCode()
- == TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
- ConfigNodeTestUtils.checkDataPartitionTable(
- storageGroup,
- j,
- j + seriesPartitionBatchSize,
- k,
- k + timePartitionBatchSize,
- testTimePartitionInterval,
- configNodeClient
- .getDataPartitionTable(dataPartitionReq)
- .getDataPartitionTable());
- break;
- }
- } catch (Exception e) {
- // Retry sometimes in order to avoid request timeout
- LOGGER.error(e.getMessage());
- TimeUnit.SECONDS.sleep(1);
+ // Test1: divide and inherit DataPartitions from scratch
+ for (long timePartitionSlot = baseStartTime;
+ timePartitionSlot < baseStartTime + testTimePartitionSlotsNum;
+ timePartitionSlot++) {
+ for (int seriesPartitionSlot = 0;
+ seriesPartitionSlot < testSeriesSlotNum;
+ seriesPartitionSlot += seriesPartitionSlotBatchSize) {
+ ConfigNodeTestUtils.getOrCreateDataPartitionWithRetry(
+ database,
+ seriesPartitionSlot,
+ seriesPartitionSlot + seriesPartitionSlotBatchSize,
+ timePartitionSlot,
+ timePartitionSlot + 1,
+ testTimePartitionInterval);
+ }
+ }
+
+ int mu = (int) (testSeriesSlotNum / testDataRegionPerDataNode);
+ TDataPartitionTableResp dataPartitionTableResp =
+ ConfigNodeTestUtils.getDataPartitionWithRetry(
+ database,
+ 0,
+ testSeriesSlotNum,
+ baseStartTime,
+ baseStartTime + testTimePartitionSlotsNum,
+ testTimePartitionInterval);
+ Assert.assertNotNull(dataPartitionTableResp);
+
+ // All DataRegionGroups divide all SeriesSlots evenly
+ final int expectedPartitionNum1 = mu * testTimePartitionSlotsNum;
+ Map<TConsensusGroupId, Integer> counter =
+ ConfigNodeTestUtils.countDataPartition(
+ dataPartitionTableResp.getDataPartitionTable().get(database));
+ counter.forEach((groupId, num) -> Assert.assertEquals(expectedPartitionNum1, num.intValue()));
+
+ // Test DataPartition inherit policy
+ dataPartitionTableResp
+ .getDataPartitionTable()
+ .get(database)
+ .forEach(
+ ((seriesPartitionSlot, timePartitionSlotMap) -> {
+ // All Timeslots belonging to the same SeriesSlot are allocated to the same
+ // DataRegionGroup
+ TConsensusGroupId groupId =
+ timePartitionSlotMap
+ .get(new TTimePartitionSlot(baseStartTime * testTimePartitionInterval))
+ .get(0);
+ timePartitionSlotMap.forEach(
+ (timePartitionSlot, groupIdList) ->
+ Assert.assertEquals(groupId, groupIdList.get(0)));
+ dataAllotTable1.put(seriesPartitionSlot, groupId);
+ }));
+
+ // Register a new DataNode to extend DataRegionGroups
+ EnvFactory.getEnv().registerNewDataNode(true);
+
+ // Test2: divide and inherit DataPartitions after extension
+ mu = (int) (testSeriesSlotNum / (testDataRegionPerDataNode * 2));
+ dataPartitionTableResp =
+ ConfigNodeTestUtils.getOrCreateDataPartitionWithRetry(
+ database,
+ 0,
+ testSeriesSlotNum,
+ baseStartTime + testTimePartitionSlotsNum,
+ baseStartTime + testTimePartitionSlotsNum + timePartitionBatchSize,
+ testTimePartitionInterval);
+ Assert.assertNotNull(dataPartitionTableResp);
+
+ // All DataRegionGroups divide all SeriesSlots evenly
+ counter =
+ ConfigNodeTestUtils.countDataPartition(
+ dataPartitionTableResp.getDataPartitionTable().get(database));
+ final int expectedPartitionNum2 = mu * timePartitionBatchSize;
+ counter.forEach((groupId, num) -> Assert.assertEquals(expectedPartitionNum2, num.intValue()));
+
+ // Test DataPartition inherit policy
+ Map<TSeriesPartitionSlot, TConsensusGroupId> dataAllotTable2 = new ConcurrentHashMap<>();
+ dataPartitionTableResp
+ .getDataPartitionTable()
+ .get(database)
+ .forEach(
+ ((seriesPartitionSlot, timePartitionSlotMap) -> {
+ // All Timeslots belonging to the same SeriesSlot are allocated to the same
+ // DataRegionGroup
+ TConsensusGroupId groupId =
+ timePartitionSlotMap
+ .get(
+ new TTimePartitionSlot(
+ (baseStartTime + testTimePartitionSlotsNum)
+ * testTimePartitionInterval))
+ .get(0);
+ timePartitionSlotMap.forEach(
+ (timePartitionSlot, groupIdList) ->
+ Assert.assertEquals(groupId, groupIdList.get(0)));
+
+ if (dataAllotTable1.containsValue(groupId)) {
+ // The DataRegionGroup has been inherited
Review Comment:
add size judgement here? Exactly half of the slots should have changed?
##########
iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/structure/BalanceTreeMap.java:
##########
@@ -0,0 +1,98 @@
+/*
+ * 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.iotdb.commons.structure;
+
+import org.apache.iotdb.commons.utils.TestOnly;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Set;
+import java.util.TreeMap;
+
+public class BalanceTreeMap<K, V extends Comparable<V>> {
+
+ private final HashMap<K, V> keyValueMap;
+ private final TreeMap<V, Set<K>> valueKeyMap;
+
+ public BalanceTreeMap() {
+ this.keyValueMap = new HashMap<>();
+ this.valueKeyMap = new TreeMap<>();
+ }
+
+ /**
+ * Put or modify a key-value pair.
+ *
+ * @param key Key
+ * @param value Value
+ */
+ public void put(K key, V value) {
+ V oldValue = keyValueMap.getOrDefault(key, null);
+
+ // Update keyValueMap
+ keyValueMap.put(key, value);
Review Comment:
you can directly use `V oldValue = keyValueMap.put(key, value);`
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/PartitionBalancer.java:
##########
@@ -16,58 +16,237 @@
* specific language governing permissions and limitations
* under the License.
*/
+
package org.apache.iotdb.confignode.manager.load.balancer;
+import org.apache.iotdb.common.rpc.thrift.TConsensusGroupId;
+import org.apache.iotdb.common.rpc.thrift.TConsensusGroupType;
import org.apache.iotdb.common.rpc.thrift.TSeriesPartitionSlot;
+import org.apache.iotdb.common.rpc.thrift.TTimePartitionSlot;
import org.apache.iotdb.commons.partition.DataPartitionTable;
import org.apache.iotdb.commons.partition.SchemaPartitionTable;
+import org.apache.iotdb.commons.partition.SeriesPartitionTable;
+import org.apache.iotdb.commons.structure.BalanceTreeMap;
+import org.apache.iotdb.confignode.exception.DatabaseNotExistsException;
import org.apache.iotdb.confignode.exception.NoAvailableRegionGroupException;
import org.apache.iotdb.confignode.manager.IManager;
-import org.apache.iotdb.confignode.manager.load.balancer.partition.GreedyPartitionAllocator;
-import org.apache.iotdb.confignode.manager.load.balancer.partition.IPartitionAllocator;
+import org.apache.iotdb.confignode.manager.load.balancer.partition.DataPartitionPolicyTable;
+import org.apache.iotdb.confignode.manager.partition.PartitionManager;
+import org.apache.iotdb.confignode.manager.schema.ClusterSchemaManager;
import org.apache.iotdb.confignode.rpc.thrift.TTimeSlotList;
+import org.apache.iotdb.tsfile.utils.Pair;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import java.util.Comparator;
import java.util.List;
import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
/**
* The SeriesPartitionSlotBalancer provides interfaces to generate optimal Partition allocation and
* migration plans
*/
public class PartitionBalancer {
+ private static final Logger LOGGER = LoggerFactory.getLogger(PartitionBalancer.class);
+
private final IManager configManager;
+ // Map<DatabaseName, DataPartitionPolicyTable>
+ private final Map<String, DataPartitionPolicyTable> dataAllotTableMap;
+
public PartitionBalancer(IManager configManager) {
this.configManager = configManager;
+ this.dataAllotTableMap = new ConcurrentHashMap<>();
}
/**
* Allocate SchemaPartitions
*
* @param unassignedSchemaPartitionSlotsMap SchemaPartitionSlots that should be assigned
- * @return Map<StorageGroupName, SchemaPartitionTable>, the allocating result
+ * @return Map<DatabaseName, SchemaPartitionTable>, the allocating result
*/
public Map<String, SchemaPartitionTable> allocateSchemaPartition(
Map<String, List<TSeriesPartitionSlot>> unassignedSchemaPartitionSlotsMap)
throws NoAvailableRegionGroupException {
- return genPartitionAllocator().allocateSchemaPartition(unassignedSchemaPartitionSlotsMap);
+ Map<String, SchemaPartitionTable> result = new ConcurrentHashMap<>();
+
+ for (Map.Entry<String, List<TSeriesPartitionSlot>> slotsMapEntry :
+ unassignedSchemaPartitionSlotsMap.entrySet()) {
+ final String database = slotsMapEntry.getKey();
+ final List<TSeriesPartitionSlot> unassignedPartitionSlots = slotsMapEntry.getValue();
+
+ // Filter available SchemaRegionGroups and
+ // sort them by the number of allocated SchemaPartitions
+ BalanceTreeMap<TConsensusGroupId, Integer> counter = new BalanceTreeMap<>();
+ List<Pair<Long, TConsensusGroupId>> regionSlotsCounter =
+ getPartitionManager()
+ .getSortedRegionGroupSlotsCounter(database, TConsensusGroupType.SchemaRegion);
+ for (Pair<Long, TConsensusGroupId> pair : regionSlotsCounter) {
+ counter.put(pair.getRight(), pair.getLeft().intValue());
+ }
+
+ // Enumerate SeriesPartitionSlot
+ Map<TSeriesPartitionSlot, TConsensusGroupId> schemaPartitionMap = new ConcurrentHashMap<>();
Review Comment:
same as above
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/PartitionBalancer.java:
##########
@@ -16,58 +16,237 @@
* specific language governing permissions and limitations
* under the License.
*/
+
package org.apache.iotdb.confignode.manager.load.balancer;
+import org.apache.iotdb.common.rpc.thrift.TConsensusGroupId;
+import org.apache.iotdb.common.rpc.thrift.TConsensusGroupType;
import org.apache.iotdb.common.rpc.thrift.TSeriesPartitionSlot;
+import org.apache.iotdb.common.rpc.thrift.TTimePartitionSlot;
import org.apache.iotdb.commons.partition.DataPartitionTable;
import org.apache.iotdb.commons.partition.SchemaPartitionTable;
+import org.apache.iotdb.commons.partition.SeriesPartitionTable;
+import org.apache.iotdb.commons.structure.BalanceTreeMap;
+import org.apache.iotdb.confignode.exception.DatabaseNotExistsException;
import org.apache.iotdb.confignode.exception.NoAvailableRegionGroupException;
import org.apache.iotdb.confignode.manager.IManager;
-import org.apache.iotdb.confignode.manager.load.balancer.partition.GreedyPartitionAllocator;
-import org.apache.iotdb.confignode.manager.load.balancer.partition.IPartitionAllocator;
+import org.apache.iotdb.confignode.manager.load.balancer.partition.DataPartitionPolicyTable;
+import org.apache.iotdb.confignode.manager.partition.PartitionManager;
+import org.apache.iotdb.confignode.manager.schema.ClusterSchemaManager;
import org.apache.iotdb.confignode.rpc.thrift.TTimeSlotList;
+import org.apache.iotdb.tsfile.utils.Pair;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import java.util.Comparator;
import java.util.List;
import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
/**
* The SeriesPartitionSlotBalancer provides interfaces to generate optimal Partition allocation and
* migration plans
*/
public class PartitionBalancer {
+ private static final Logger LOGGER = LoggerFactory.getLogger(PartitionBalancer.class);
+
private final IManager configManager;
+ // Map<DatabaseName, DataPartitionPolicyTable>
+ private final Map<String, DataPartitionPolicyTable> dataAllotTableMap;
+
public PartitionBalancer(IManager configManager) {
this.configManager = configManager;
+ this.dataAllotTableMap = new ConcurrentHashMap<>();
}
/**
* Allocate SchemaPartitions
*
* @param unassignedSchemaPartitionSlotsMap SchemaPartitionSlots that should be assigned
- * @return Map<StorageGroupName, SchemaPartitionTable>, the allocating result
+ * @return Map<DatabaseName, SchemaPartitionTable>, the allocating result
*/
public Map<String, SchemaPartitionTable> allocateSchemaPartition(
Map<String, List<TSeriesPartitionSlot>> unassignedSchemaPartitionSlotsMap)
throws NoAvailableRegionGroupException {
- return genPartitionAllocator().allocateSchemaPartition(unassignedSchemaPartitionSlotsMap);
+ Map<String, SchemaPartitionTable> result = new ConcurrentHashMap<>();
Review Comment:
Do we really need `ConcurrentHashMap`?
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/PartitionBalancer.java:
##########
@@ -16,58 +16,237 @@
* specific language governing permissions and limitations
* under the License.
*/
+
package org.apache.iotdb.confignode.manager.load.balancer;
+import org.apache.iotdb.common.rpc.thrift.TConsensusGroupId;
+import org.apache.iotdb.common.rpc.thrift.TConsensusGroupType;
import org.apache.iotdb.common.rpc.thrift.TSeriesPartitionSlot;
+import org.apache.iotdb.common.rpc.thrift.TTimePartitionSlot;
import org.apache.iotdb.commons.partition.DataPartitionTable;
import org.apache.iotdb.commons.partition.SchemaPartitionTable;
+import org.apache.iotdb.commons.partition.SeriesPartitionTable;
+import org.apache.iotdb.commons.structure.BalanceTreeMap;
+import org.apache.iotdb.confignode.exception.DatabaseNotExistsException;
import org.apache.iotdb.confignode.exception.NoAvailableRegionGroupException;
import org.apache.iotdb.confignode.manager.IManager;
-import org.apache.iotdb.confignode.manager.load.balancer.partition.GreedyPartitionAllocator;
-import org.apache.iotdb.confignode.manager.load.balancer.partition.IPartitionAllocator;
+import org.apache.iotdb.confignode.manager.load.balancer.partition.DataPartitionPolicyTable;
+import org.apache.iotdb.confignode.manager.partition.PartitionManager;
+import org.apache.iotdb.confignode.manager.schema.ClusterSchemaManager;
import org.apache.iotdb.confignode.rpc.thrift.TTimeSlotList;
+import org.apache.iotdb.tsfile.utils.Pair;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import java.util.Comparator;
import java.util.List;
import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
/**
* The SeriesPartitionSlotBalancer provides interfaces to generate optimal Partition allocation and
* migration plans
*/
public class PartitionBalancer {
+ private static final Logger LOGGER = LoggerFactory.getLogger(PartitionBalancer.class);
+
private final IManager configManager;
+ // Map<DatabaseName, DataPartitionPolicyTable>
+ private final Map<String, DataPartitionPolicyTable> dataAllotTableMap;
+
public PartitionBalancer(IManager configManager) {
this.configManager = configManager;
+ this.dataAllotTableMap = new ConcurrentHashMap<>();
}
/**
* Allocate SchemaPartitions
*
* @param unassignedSchemaPartitionSlotsMap SchemaPartitionSlots that should be assigned
- * @return Map<StorageGroupName, SchemaPartitionTable>, the allocating result
+ * @return Map<DatabaseName, SchemaPartitionTable>, the allocating result
*/
public Map<String, SchemaPartitionTable> allocateSchemaPartition(
Map<String, List<TSeriesPartitionSlot>> unassignedSchemaPartitionSlotsMap)
throws NoAvailableRegionGroupException {
- return genPartitionAllocator().allocateSchemaPartition(unassignedSchemaPartitionSlotsMap);
+ Map<String, SchemaPartitionTable> result = new ConcurrentHashMap<>();
+
+ for (Map.Entry<String, List<TSeriesPartitionSlot>> slotsMapEntry :
+ unassignedSchemaPartitionSlotsMap.entrySet()) {
+ final String database = slotsMapEntry.getKey();
+ final List<TSeriesPartitionSlot> unassignedPartitionSlots = slotsMapEntry.getValue();
+
+ // Filter available SchemaRegionGroups and
+ // sort them by the number of allocated SchemaPartitions
+ BalanceTreeMap<TConsensusGroupId, Integer> counter = new BalanceTreeMap<>();
+ List<Pair<Long, TConsensusGroupId>> regionSlotsCounter =
+ getPartitionManager()
+ .getSortedRegionGroupSlotsCounter(database, TConsensusGroupType.SchemaRegion);
+ for (Pair<Long, TConsensusGroupId> pair : regionSlotsCounter) {
+ counter.put(pair.getRight(), pair.getLeft().intValue());
+ }
+
+ // Enumerate SeriesPartitionSlot
+ Map<TSeriesPartitionSlot, TConsensusGroupId> schemaPartitionMap = new ConcurrentHashMap<>();
+ for (TSeriesPartitionSlot seriesPartitionSlot : unassignedPartitionSlots) {
+ // Greedy allocation: allocate the unassigned SchemaPartition to
+ // the RegionGroup whose allocated SchemaPartitions is the least
+ TConsensusGroupId consensusGroupId = counter.getKeyWithMinValue();
+ schemaPartitionMap.put(seriesPartitionSlot, consensusGroupId);
+ counter.put(consensusGroupId, counter.get(consensusGroupId) + 1);
+ }
+ result.put(database, new SchemaPartitionTable(schemaPartitionMap));
+ }
+
+ return result;
}
/**
* Allocate DataPartitions
*
* @param unassignedDataPartitionSlotsMap DataPartitionSlots that should be assigned
- * @return Map<StorageGroupName, DataPartitionTable>, the allocating result
+ * @return Map<DatabaseName, DataPartitionTable>, the allocating result
*/
public Map<String, DataPartitionTable> allocateDataPartition(
Map<String, Map<TSeriesPartitionSlot, TTimeSlotList>> unassignedDataPartitionSlotsMap)
throws NoAvailableRegionGroupException {
- return genPartitionAllocator().allocateDataPartition(unassignedDataPartitionSlotsMap);
+ Map<String, DataPartitionTable> result = new ConcurrentHashMap<>();
+
+ for (Map.Entry<String, Map<TSeriesPartitionSlot, TTimeSlotList>> slotsMapEntry :
+ unassignedDataPartitionSlotsMap.entrySet()) {
+ final String database = slotsMapEntry.getKey();
+ final Map<TSeriesPartitionSlot, TTimeSlotList> unassignedPartitionSlotsMap =
+ slotsMapEntry.getValue();
+
+ // Filter available DataRegionGroups and
+ // sort them by the number of allocated DataPartitions
+ BalanceTreeMap<TConsensusGroupId, Integer> counter = new BalanceTreeMap<>();
+ List<Pair<Long, TConsensusGroupId>> regionSlotsCounter =
+ getPartitionManager()
+ .getSortedRegionGroupSlotsCounter(database, TConsensusGroupType.DataRegion);
+ for (Pair<Long, TConsensusGroupId> pair : regionSlotsCounter) {
+ counter.put(pair.getRight(), pair.getLeft().intValue());
+ }
+
+ DataPartitionPolicyTable allotTable = dataAllotTableMap.get(database);
+ allotTable.acquireLock();
Review Comment:
put this in try scope ?
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/PartitionBalancer.java:
##########
@@ -16,58 +16,237 @@
* specific language governing permissions and limitations
* under the License.
*/
+
package org.apache.iotdb.confignode.manager.load.balancer;
+import org.apache.iotdb.common.rpc.thrift.TConsensusGroupId;
+import org.apache.iotdb.common.rpc.thrift.TConsensusGroupType;
import org.apache.iotdb.common.rpc.thrift.TSeriesPartitionSlot;
+import org.apache.iotdb.common.rpc.thrift.TTimePartitionSlot;
import org.apache.iotdb.commons.partition.DataPartitionTable;
import org.apache.iotdb.commons.partition.SchemaPartitionTable;
+import org.apache.iotdb.commons.partition.SeriesPartitionTable;
+import org.apache.iotdb.commons.structure.BalanceTreeMap;
+import org.apache.iotdb.confignode.exception.DatabaseNotExistsException;
import org.apache.iotdb.confignode.exception.NoAvailableRegionGroupException;
import org.apache.iotdb.confignode.manager.IManager;
-import org.apache.iotdb.confignode.manager.load.balancer.partition.GreedyPartitionAllocator;
-import org.apache.iotdb.confignode.manager.load.balancer.partition.IPartitionAllocator;
+import org.apache.iotdb.confignode.manager.load.balancer.partition.DataPartitionPolicyTable;
+import org.apache.iotdb.confignode.manager.partition.PartitionManager;
+import org.apache.iotdb.confignode.manager.schema.ClusterSchemaManager;
import org.apache.iotdb.confignode.rpc.thrift.TTimeSlotList;
+import org.apache.iotdb.tsfile.utils.Pair;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import java.util.Comparator;
import java.util.List;
import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
/**
* The SeriesPartitionSlotBalancer provides interfaces to generate optimal Partition allocation and
* migration plans
*/
public class PartitionBalancer {
+ private static final Logger LOGGER = LoggerFactory.getLogger(PartitionBalancer.class);
+
private final IManager configManager;
+ // Map<DatabaseName, DataPartitionPolicyTable>
+ private final Map<String, DataPartitionPolicyTable> dataAllotTableMap;
+
public PartitionBalancer(IManager configManager) {
this.configManager = configManager;
+ this.dataAllotTableMap = new ConcurrentHashMap<>();
}
/**
* Allocate SchemaPartitions
*
* @param unassignedSchemaPartitionSlotsMap SchemaPartitionSlots that should be assigned
- * @return Map<StorageGroupName, SchemaPartitionTable>, the allocating result
+ * @return Map<DatabaseName, SchemaPartitionTable>, the allocating result
*/
public Map<String, SchemaPartitionTable> allocateSchemaPartition(
Map<String, List<TSeriesPartitionSlot>> unassignedSchemaPartitionSlotsMap)
throws NoAvailableRegionGroupException {
- return genPartitionAllocator().allocateSchemaPartition(unassignedSchemaPartitionSlotsMap);
+ Map<String, SchemaPartitionTable> result = new ConcurrentHashMap<>();
+
+ for (Map.Entry<String, List<TSeriesPartitionSlot>> slotsMapEntry :
+ unassignedSchemaPartitionSlotsMap.entrySet()) {
+ final String database = slotsMapEntry.getKey();
+ final List<TSeriesPartitionSlot> unassignedPartitionSlots = slotsMapEntry.getValue();
+
+ // Filter available SchemaRegionGroups and
+ // sort them by the number of allocated SchemaPartitions
+ BalanceTreeMap<TConsensusGroupId, Integer> counter = new BalanceTreeMap<>();
+ List<Pair<Long, TConsensusGroupId>> regionSlotsCounter =
+ getPartitionManager()
+ .getSortedRegionGroupSlotsCounter(database, TConsensusGroupType.SchemaRegion);
+ for (Pair<Long, TConsensusGroupId> pair : regionSlotsCounter) {
+ counter.put(pair.getRight(), pair.getLeft().intValue());
+ }
+
+ // Enumerate SeriesPartitionSlot
+ Map<TSeriesPartitionSlot, TConsensusGroupId> schemaPartitionMap = new ConcurrentHashMap<>();
+ for (TSeriesPartitionSlot seriesPartitionSlot : unassignedPartitionSlots) {
+ // Greedy allocation: allocate the unassigned SchemaPartition to
+ // the RegionGroup whose allocated SchemaPartitions is the least
+ TConsensusGroupId consensusGroupId = counter.getKeyWithMinValue();
+ schemaPartitionMap.put(seriesPartitionSlot, consensusGroupId);
+ counter.put(consensusGroupId, counter.get(consensusGroupId) + 1);
+ }
+ result.put(database, new SchemaPartitionTable(schemaPartitionMap));
+ }
+
+ return result;
}
/**
* Allocate DataPartitions
*
* @param unassignedDataPartitionSlotsMap DataPartitionSlots that should be assigned
- * @return Map<StorageGroupName, DataPartitionTable>, the allocating result
+ * @return Map<DatabaseName, DataPartitionTable>, the allocating result
*/
public Map<String, DataPartitionTable> allocateDataPartition(
Map<String, Map<TSeriesPartitionSlot, TTimeSlotList>> unassignedDataPartitionSlotsMap)
throws NoAvailableRegionGroupException {
- return genPartitionAllocator().allocateDataPartition(unassignedDataPartitionSlotsMap);
+ Map<String, DataPartitionTable> result = new ConcurrentHashMap<>();
Review Comment:
same as above
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/partition/PartitionInfo.java:
##########
@@ -784,6 +822,19 @@ public Set<Integer> getAllSchemaPartition() {
return schemaPartitionSet;
}
+ /**
+ * Get the last DataAllotTable of the specified Database.
+ *
+ * @param database The specified Database
+ * @return The last DataAllotTable
+ */
+ public Map<TSeriesPartitionSlot, TConsensusGroupId> getLastDataAllotTable(String database) {
+ if (isDatabaseExisted(database)) {
+ return databasePartitionTables.get(database).getLastDataAllotTable();
+ }
+ return new HashMap<>();
Review Comment:
Collections.emptyMap()
##########
integration-test/src/test/java/org/apache/iotdb/confignode/it/partition/IoTDBPartitionCreationIT.java:
##########
@@ -419,182 +419,6 @@ public void testPartitionAllocation() throws Exception {
4 + testTimePartitionBatchSize,
testTimePartitionInterval,
dataPartitionTableResp.getDataPartitionTable());
-
Review Comment:
Why we remove these ITs? Do we still have IT in the exception state of region
##########
iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/partition/DataPartitionEntry.java:
##########
@@ -0,0 +1,87 @@
+/*
+ * 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.iotdb.commons.partition;
+
+import org.apache.iotdb.common.rpc.thrift.TConsensusGroupId;
+import org.apache.iotdb.common.rpc.thrift.TSeriesPartitionSlot;
+import org.apache.iotdb.common.rpc.thrift.TTimePartitionSlot;
+
+import java.security.SecureRandom;
+import java.util.Objects;
+
+public class DataPartitionEntry implements Comparable<DataPartitionEntry> {
Review Comment:
It doesn't look like this class is being used, so why does it exist?
##########
iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/structure/BalanceTreeMap.java:
##########
@@ -0,0 +1,98 @@
+/*
+ * 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.iotdb.commons.structure;
+
+import org.apache.iotdb.commons.utils.TestOnly;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Set;
+import java.util.TreeMap;
+
+public class BalanceTreeMap<K, V extends Comparable<V>> {
+
+ private final HashMap<K, V> keyValueMap;
+ private final TreeMap<V, Set<K>> valueKeyMap;
+
+ public BalanceTreeMap() {
+ this.keyValueMap = new HashMap<>();
+ this.valueKeyMap = new TreeMap<>();
+ }
+
+ /**
+ * Put or modify a key-value pair.
+ *
+ * @param key Key
+ * @param value Value
+ */
+ public void put(K key, V value) {
+ V oldValue = keyValueMap.getOrDefault(key, null);
+
+ // Update keyValueMap
+ keyValueMap.put(key, value);
+
+ // Update valueKeyMap
+ if (oldValue != null) {
+ valueKeyMap.get(oldValue).remove(key);
+ if (valueKeyMap.get(oldValue).isEmpty()) {
+ valueKeyMap.remove(oldValue);
+ }
+ }
+ valueKeyMap.computeIfAbsent(value, empty -> new HashSet<>()).add(key);
+ }
+
+ /**
+ * Get key with minimum value.
+ *
+ * @return Key with minimum value
+ */
+ public K getKeyWithMinValue() {
+ return valueKeyMap.firstEntry().getValue().iterator().next();
+ }
+
+ public V get(K key) {
+ return keyValueMap.getOrDefault(key, null);
+ }
+
+ public boolean containsKey(K key) {
+ return keyValueMap.containsKey(key);
+ }
+
+ public int size() {
+ return keyValueMap.size();
+ }
+
+ @TestOnly
+ public void remove(K key) {
Review Comment:
```
@TestOnly
public void remove(K key) {
V value = keyValueMap.remove(key);
if (value != null) {
Set<K> keysSet = valueKeyMap.get(value);
if (keysSet != null) {
keysSet.remove(key);
if (keysSet.isEmpty()) {
valueKeyMap.remove(value);
}
}
}
}
```
##########
integration-test/src/test/java/org/apache/iotdb/confignode/it/partition/IoTDBPartitionInheritPolicyIT.java:
##########
@@ -95,76 +91,173 @@ public static void tearDown() {
@Test
public void testDataPartitionInheritPolicy() throws Exception {
+ final long baseStartTime = 1000;
+ Map<TSeriesPartitionSlot, TConsensusGroupId> dataAllotTable1 = new ConcurrentHashMap<>();
- try (SyncConfigNodeIServiceClient client =
- (SyncConfigNodeIServiceClient) EnvFactory.getEnv().getLeaderConfigNodeConnection()) {
- TDataPartitionReq dataPartitionReq = new TDataPartitionReq();
- TDataPartitionTableResp dataPartitionTableResp;
- Map<String, Map<TSeriesPartitionSlot, TTimeSlotList>> partitionSlotsMap;
-
- for (int i = 0; i < storageGroupNum; i++) {
- String storageGroup = sg + i;
- for (int j = 0; j < testSeriesPartitionSlotNum; j += seriesPartitionBatchSize) {
- // Test inherit predecessor or successor
- boolean isAscending = (j / 10) % 2 == 0;
- int step = isAscending ? timePartitionBatchSize : -timePartitionBatchSize;
- int k = isAscending ? 0 : testTimePartitionSlotsNum - timePartitionBatchSize;
- while (0 <= k && k < testTimePartitionSlotsNum) {
- partitionSlotsMap =
- ConfigNodeTestUtils.constructPartitionSlotsMap(
- storageGroup,
- j,
- j + seriesPartitionBatchSize,
- k,
- k + timePartitionBatchSize,
- testTimePartitionInterval);
- // Let ConfigNode create DataPartition
- dataPartitionReq.setPartitionSlotsMap(partitionSlotsMap);
- for (int retry = 0; retry < 5; retry++) {
- // Build new Client since it's unstable
- try (SyncConfigNodeIServiceClient configNodeClient =
- (SyncConfigNodeIServiceClient)
- EnvFactory.getEnv().getLeaderConfigNodeConnection()) {
- dataPartitionTableResp =
- configNodeClient.getOrCreateDataPartitionTable(dataPartitionReq);
- if (dataPartitionTableResp != null
- && dataPartitionTableResp.getStatus().getCode()
- == TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
- ConfigNodeTestUtils.checkDataPartitionTable(
- storageGroup,
- j,
- j + seriesPartitionBatchSize,
- k,
- k + timePartitionBatchSize,
- testTimePartitionInterval,
- configNodeClient
- .getDataPartitionTable(dataPartitionReq)
- .getDataPartitionTable());
- break;
- }
- } catch (Exception e) {
- // Retry sometimes in order to avoid request timeout
- LOGGER.error(e.getMessage());
- TimeUnit.SECONDS.sleep(1);
+ // Test1: divide and inherit DataPartitions from scratch
+ for (long timePartitionSlot = baseStartTime;
+ timePartitionSlot < baseStartTime + testTimePartitionSlotsNum;
+ timePartitionSlot++) {
+ for (int seriesPartitionSlot = 0;
+ seriesPartitionSlot < testSeriesSlotNum;
+ seriesPartitionSlot += seriesPartitionSlotBatchSize) {
+ ConfigNodeTestUtils.getOrCreateDataPartitionWithRetry(
+ database,
+ seriesPartitionSlot,
+ seriesPartitionSlot + seriesPartitionSlotBatchSize,
+ timePartitionSlot,
+ timePartitionSlot + 1,
+ testTimePartitionInterval);
+ }
+ }
+
+ int mu = (int) (testSeriesSlotNum / testDataRegionPerDataNode);
+ TDataPartitionTableResp dataPartitionTableResp =
+ ConfigNodeTestUtils.getDataPartitionWithRetry(
+ database,
+ 0,
+ testSeriesSlotNum,
+ baseStartTime,
+ baseStartTime + testTimePartitionSlotsNum,
+ testTimePartitionInterval);
+ Assert.assertNotNull(dataPartitionTableResp);
+
+ // All DataRegionGroups divide all SeriesSlots evenly
+ final int expectedPartitionNum1 = mu * testTimePartitionSlotsNum;
+ Map<TConsensusGroupId, Integer> counter =
+ ConfigNodeTestUtils.countDataPartition(
+ dataPartitionTableResp.getDataPartitionTable().get(database));
+ counter.forEach((groupId, num) -> Assert.assertEquals(expectedPartitionNum1, num.intValue()));
+
+ // Test DataPartition inherit policy
+ dataPartitionTableResp
+ .getDataPartitionTable()
+ .get(database)
+ .forEach(
+ ((seriesPartitionSlot, timePartitionSlotMap) -> {
+ // All Timeslots belonging to the same SeriesSlot are allocated to the same
+ // DataRegionGroup
+ TConsensusGroupId groupId =
+ timePartitionSlotMap
+ .get(new TTimePartitionSlot(baseStartTime * testTimePartitionInterval))
+ .get(0);
+ timePartitionSlotMap.forEach(
+ (timePartitionSlot, groupIdList) ->
+ Assert.assertEquals(groupId, groupIdList.get(0)));
+ dataAllotTable1.put(seriesPartitionSlot, groupId);
+ }));
+
+ // Register a new DataNode to extend DataRegionGroups
+ EnvFactory.getEnv().registerNewDataNode(true);
+
+ // Test2: divide and inherit DataPartitions after extension
+ mu = (int) (testSeriesSlotNum / (testDataRegionPerDataNode * 2));
+ dataPartitionTableResp =
+ ConfigNodeTestUtils.getOrCreateDataPartitionWithRetry(
+ database,
+ 0,
+ testSeriesSlotNum,
+ baseStartTime + testTimePartitionSlotsNum,
+ baseStartTime + testTimePartitionSlotsNum + timePartitionBatchSize,
+ testTimePartitionInterval);
+ Assert.assertNotNull(dataPartitionTableResp);
+
+ // All DataRegionGroups divide all SeriesSlots evenly
+ counter =
+ ConfigNodeTestUtils.countDataPartition(
+ dataPartitionTableResp.getDataPartitionTable().get(database));
+ final int expectedPartitionNum2 = mu * timePartitionBatchSize;
+ counter.forEach((groupId, num) -> Assert.assertEquals(expectedPartitionNum2, num.intValue()));
+
+ // Test DataPartition inherit policy
+ Map<TSeriesPartitionSlot, TConsensusGroupId> dataAllotTable2 = new ConcurrentHashMap<>();
+ dataPartitionTableResp
+ .getDataPartitionTable()
+ .get(database)
+ .forEach(
+ ((seriesPartitionSlot, timePartitionSlotMap) -> {
+ // All Timeslots belonging to the same SeriesSlot are allocated to the same
+ // DataRegionGroup
+ TConsensusGroupId groupId =
+ timePartitionSlotMap
+ .get(
+ new TTimePartitionSlot(
+ (baseStartTime + testTimePartitionSlotsNum)
+ * testTimePartitionInterval))
+ .get(0);
+ timePartitionSlotMap.forEach(
+ (timePartitionSlot, groupIdList) ->
+ Assert.assertEquals(groupId, groupIdList.get(0)));
+
+ if (dataAllotTable1.containsValue(groupId)) {
+ // The DataRegionGroup has been inherited
+ Assert.assertTrue(dataAllotTable1.containsKey(seriesPartitionSlot));
+ Assert.assertEquals(dataAllotTable1.get(seriesPartitionSlot), groupId);
}
- }
- k += step;
- }
- }
+ dataAllotTable2.put(seriesPartitionSlot, groupId);
+ }));
+
+ // Test3: historical DataPartitions will inherit successor
+ Random random = new Random();
+ Set<Integer> allocatedSlots = new HashSet<>();
+ for (int i = 0; i < 10; i++) {
+ int slot = random.nextInt(testSeriesSlotNum);
+ while (allocatedSlots.contains(slot)) {
+ slot = random.nextInt(testSeriesSlotNum);
+ }
+ allocatedSlots.add(slot);
+ dataPartitionTableResp =
+ ConfigNodeTestUtils.getOrCreateDataPartitionWithRetry(
+ database,
+ slot,
+ slot + 1,
+ baseStartTime - 1,
+ baseStartTime,
+ testTimePartitionInterval);
+ Assert.assertNotNull(dataPartitionTableResp);
+
+ TSeriesPartitionSlot seriesPartitionSlot = new TSeriesPartitionSlot(slot);
+ TTimePartitionSlot timePartitionSlot =
+ new TTimePartitionSlot((baseStartTime - 1) * testTimePartitionInterval);
+ Assert.assertEquals(
+ dataAllotTable1.get(seriesPartitionSlot),
+ dataPartitionTableResp
+ .getDataPartitionTable()
+ .get(database)
+ .get(seriesPartitionSlot)
+ .get(timePartitionSlot)
+ .get(0));
+ }
+
+ // Test4: future DataPartitions will inherit predecessor
+ allocatedSlots.clear();
+ for (int i = 0; i < 10; i++) {
+ int slot = random.nextInt(testSeriesSlotNum);
+ while (allocatedSlots.contains(slot)) {
+ slot = random.nextInt(testSeriesSlotNum);
}
+ allocatedSlots.add(slot);
+ dataPartitionTableResp =
+ ConfigNodeTestUtils.getOrCreateDataPartitionWithRetry(
+ database,
+ slot,
+ slot + 1,
+ baseStartTime + 999,
+ baseStartTime + 1000,
+ testTimePartitionInterval);
+ Assert.assertNotNull(dataPartitionTableResp);
- // Test DataPartition inherit policy
- TShowRegionResp showRegionResp = client.showRegion(new TShowRegionReq());
- showRegionResp
- .getRegionInfoList()
- .forEach(
- regionInfo -> {
- // All Timeslots belonging to the same SeriesSlot are allocated to the same
- // DataRegionGroup
- Assert.assertEquals(
- regionInfo.getSeriesSlots() * testTimePartitionSlotsNum,
- regionInfo.getTimeSlots());
- });
+ TSeriesPartitionSlot seriesPartitionSlot = new TSeriesPartitionSlot(slot);
+ TTimePartitionSlot timePartitionSlot =
+ new TTimePartitionSlot((baseStartTime + 999) * testTimePartitionInterval);
Review Comment:
same
##########
iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/structure/BalanceTreeMap.java:
##########
@@ -0,0 +1,98 @@
+/*
+ * 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.iotdb.commons.structure;
+
+import org.apache.iotdb.commons.utils.TestOnly;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Set;
+import java.util.TreeMap;
+
+public class BalanceTreeMap<K, V extends Comparable<V>> {
+
+ private final HashMap<K, V> keyValueMap;
+ private final TreeMap<V, Set<K>> valueKeyMap;
+
+ public BalanceTreeMap() {
+ this.keyValueMap = new HashMap<>();
+ this.valueKeyMap = new TreeMap<>();
+ }
+
+ /**
+ * Put or modify a key-value pair.
+ *
+ * @param key Key
+ * @param value Value
+ */
+ public void put(K key, V value) {
+ V oldValue = keyValueMap.getOrDefault(key, null);
+
+ // Update keyValueMap
+ keyValueMap.put(key, value);
+
+ // Update valueKeyMap
+ if (oldValue != null) {
Review Comment:
```
Set<KeyType> keysSet = valueKeyMap.get(oldValue);
keysSet.remove(key);
if (keysSet.isEmpty()) {
valueKeyMap.remove(oldValue);
}
```
##########
integration-test/src/test/java/org/apache/iotdb/confignode/it/partition/IoTDBPartitionInheritPolicyIT.java:
##########
@@ -95,76 +91,173 @@ public static void tearDown() {
@Test
public void testDataPartitionInheritPolicy() throws Exception {
+ final long baseStartTime = 1000;
+ Map<TSeriesPartitionSlot, TConsensusGroupId> dataAllotTable1 = new ConcurrentHashMap<>();
- try (SyncConfigNodeIServiceClient client =
- (SyncConfigNodeIServiceClient) EnvFactory.getEnv().getLeaderConfigNodeConnection()) {
- TDataPartitionReq dataPartitionReq = new TDataPartitionReq();
- TDataPartitionTableResp dataPartitionTableResp;
- Map<String, Map<TSeriesPartitionSlot, TTimeSlotList>> partitionSlotsMap;
-
- for (int i = 0; i < storageGroupNum; i++) {
- String storageGroup = sg + i;
- for (int j = 0; j < testSeriesPartitionSlotNum; j += seriesPartitionBatchSize) {
- // Test inherit predecessor or successor
- boolean isAscending = (j / 10) % 2 == 0;
- int step = isAscending ? timePartitionBatchSize : -timePartitionBatchSize;
- int k = isAscending ? 0 : testTimePartitionSlotsNum - timePartitionBatchSize;
- while (0 <= k && k < testTimePartitionSlotsNum) {
- partitionSlotsMap =
- ConfigNodeTestUtils.constructPartitionSlotsMap(
- storageGroup,
- j,
- j + seriesPartitionBatchSize,
- k,
- k + timePartitionBatchSize,
- testTimePartitionInterval);
- // Let ConfigNode create DataPartition
- dataPartitionReq.setPartitionSlotsMap(partitionSlotsMap);
- for (int retry = 0; retry < 5; retry++) {
- // Build new Client since it's unstable
- try (SyncConfigNodeIServiceClient configNodeClient =
- (SyncConfigNodeIServiceClient)
- EnvFactory.getEnv().getLeaderConfigNodeConnection()) {
- dataPartitionTableResp =
- configNodeClient.getOrCreateDataPartitionTable(dataPartitionReq);
- if (dataPartitionTableResp != null
- && dataPartitionTableResp.getStatus().getCode()
- == TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
- ConfigNodeTestUtils.checkDataPartitionTable(
- storageGroup,
- j,
- j + seriesPartitionBatchSize,
- k,
- k + timePartitionBatchSize,
- testTimePartitionInterval,
- configNodeClient
- .getDataPartitionTable(dataPartitionReq)
- .getDataPartitionTable());
- break;
- }
- } catch (Exception e) {
- // Retry sometimes in order to avoid request timeout
- LOGGER.error(e.getMessage());
- TimeUnit.SECONDS.sleep(1);
+ // Test1: divide and inherit DataPartitions from scratch
+ for (long timePartitionSlot = baseStartTime;
+ timePartitionSlot < baseStartTime + testTimePartitionSlotsNum;
+ timePartitionSlot++) {
+ for (int seriesPartitionSlot = 0;
+ seriesPartitionSlot < testSeriesSlotNum;
+ seriesPartitionSlot += seriesPartitionSlotBatchSize) {
+ ConfigNodeTestUtils.getOrCreateDataPartitionWithRetry(
+ database,
+ seriesPartitionSlot,
+ seriesPartitionSlot + seriesPartitionSlotBatchSize,
+ timePartitionSlot,
+ timePartitionSlot + 1,
+ testTimePartitionInterval);
+ }
+ }
+
+ int mu = (int) (testSeriesSlotNum / testDataRegionPerDataNode);
+ TDataPartitionTableResp dataPartitionTableResp =
+ ConfigNodeTestUtils.getDataPartitionWithRetry(
+ database,
+ 0,
+ testSeriesSlotNum,
+ baseStartTime,
+ baseStartTime + testTimePartitionSlotsNum,
+ testTimePartitionInterval);
+ Assert.assertNotNull(dataPartitionTableResp);
+
+ // All DataRegionGroups divide all SeriesSlots evenly
+ final int expectedPartitionNum1 = mu * testTimePartitionSlotsNum;
+ Map<TConsensusGroupId, Integer> counter =
+ ConfigNodeTestUtils.countDataPartition(
+ dataPartitionTableResp.getDataPartitionTable().get(database));
+ counter.forEach((groupId, num) -> Assert.assertEquals(expectedPartitionNum1, num.intValue()));
+
+ // Test DataPartition inherit policy
+ dataPartitionTableResp
+ .getDataPartitionTable()
+ .get(database)
+ .forEach(
+ ((seriesPartitionSlot, timePartitionSlotMap) -> {
+ // All Timeslots belonging to the same SeriesSlot are allocated to the same
+ // DataRegionGroup
+ TConsensusGroupId groupId =
+ timePartitionSlotMap
+ .get(new TTimePartitionSlot(baseStartTime * testTimePartitionInterval))
+ .get(0);
+ timePartitionSlotMap.forEach(
+ (timePartitionSlot, groupIdList) ->
+ Assert.assertEquals(groupId, groupIdList.get(0)));
+ dataAllotTable1.put(seriesPartitionSlot, groupId);
+ }));
+
+ // Register a new DataNode to extend DataRegionGroups
+ EnvFactory.getEnv().registerNewDataNode(true);
+
+ // Test2: divide and inherit DataPartitions after extension
+ mu = (int) (testSeriesSlotNum / (testDataRegionPerDataNode * 2));
+ dataPartitionTableResp =
+ ConfigNodeTestUtils.getOrCreateDataPartitionWithRetry(
+ database,
+ 0,
+ testSeriesSlotNum,
+ baseStartTime + testTimePartitionSlotsNum,
+ baseStartTime + testTimePartitionSlotsNum + timePartitionBatchSize,
+ testTimePartitionInterval);
+ Assert.assertNotNull(dataPartitionTableResp);
+
+ // All DataRegionGroups divide all SeriesSlots evenly
+ counter =
+ ConfigNodeTestUtils.countDataPartition(
+ dataPartitionTableResp.getDataPartitionTable().get(database));
+ final int expectedPartitionNum2 = mu * timePartitionBatchSize;
+ counter.forEach((groupId, num) -> Assert.assertEquals(expectedPartitionNum2, num.intValue()));
+
+ // Test DataPartition inherit policy
+ Map<TSeriesPartitionSlot, TConsensusGroupId> dataAllotTable2 = new ConcurrentHashMap<>();
+ dataPartitionTableResp
+ .getDataPartitionTable()
+ .get(database)
+ .forEach(
+ ((seriesPartitionSlot, timePartitionSlotMap) -> {
+ // All Timeslots belonging to the same SeriesSlot are allocated to the same
+ // DataRegionGroup
+ TConsensusGroupId groupId =
+ timePartitionSlotMap
+ .get(
+ new TTimePartitionSlot(
+ (baseStartTime + testTimePartitionSlotsNum)
+ * testTimePartitionInterval))
+ .get(0);
+ timePartitionSlotMap.forEach(
+ (timePartitionSlot, groupIdList) ->
+ Assert.assertEquals(groupId, groupIdList.get(0)));
+
+ if (dataAllotTable1.containsValue(groupId)) {
+ // The DataRegionGroup has been inherited
+ Assert.assertTrue(dataAllotTable1.containsKey(seriesPartitionSlot));
+ Assert.assertEquals(dataAllotTable1.get(seriesPartitionSlot), groupId);
}
- }
- k += step;
- }
- }
+ dataAllotTable2.put(seriesPartitionSlot, groupId);
+ }));
+
+ // Test3: historical DataPartitions will inherit successor
+ Random random = new Random();
+ Set<Integer> allocatedSlots = new HashSet<>();
+ for (int i = 0; i < 10; i++) {
+ int slot = random.nextInt(testSeriesSlotNum);
+ while (allocatedSlots.contains(slot)) {
+ slot = random.nextInt(testSeriesSlotNum);
+ }
+ allocatedSlots.add(slot);
+ dataPartitionTableResp =
+ ConfigNodeTestUtils.getOrCreateDataPartitionWithRetry(
+ database,
+ slot,
+ slot + 1,
+ baseStartTime - 1,
+ baseStartTime,
+ testTimePartitionInterval);
+ Assert.assertNotNull(dataPartitionTableResp);
+
+ TSeriesPartitionSlot seriesPartitionSlot = new TSeriesPartitionSlot(slot);
+ TTimePartitionSlot timePartitionSlot =
+ new TTimePartitionSlot((baseStartTime - 1) * testTimePartitionInterval);
+ Assert.assertEquals(
+ dataAllotTable1.get(seriesPartitionSlot),
+ dataPartitionTableResp
+ .getDataPartitionTable()
+ .get(database)
+ .get(seriesPartitionSlot)
+ .get(timePartitionSlot)
+ .get(0));
+ }
+
+ // Test4: future DataPartitions will inherit predecessor
+ allocatedSlots.clear();
+ for (int i = 0; i < 10; i++) {
+ int slot = random.nextInt(testSeriesSlotNum);
+ while (allocatedSlots.contains(slot)) {
+ slot = random.nextInt(testSeriesSlotNum);
}
+ allocatedSlots.add(slot);
+ dataPartitionTableResp =
+ ConfigNodeTestUtils.getOrCreateDataPartitionWithRetry(
+ database,
+ slot,
+ slot + 1,
+ baseStartTime + 999,
Review Comment:
do not use magic number, you can use `testTimePartitionSlotsNum * 10 - 1` and `testTimePartitionSlotsNum * 10`
##########
iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/structure/BalanceTreeMap.java:
##########
@@ -0,0 +1,98 @@
+/*
+ * 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.iotdb.commons.structure;
+
+import org.apache.iotdb.commons.utils.TestOnly;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Set;
+import java.util.TreeMap;
+
+public class BalanceTreeMap<K, V extends Comparable<V>> {
Review Comment:
Add some comments for this structure
##########
iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/structure/BalanceTreeMap.java:
##########
@@ -0,0 +1,98 @@
+/*
+ * 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.iotdb.commons.structure;
+
+import org.apache.iotdb.commons.utils.TestOnly;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Set;
+import java.util.TreeMap;
+
+public class BalanceTreeMap<K, V extends Comparable<V>> {
+
+ private final HashMap<K, V> keyValueMap;
+ private final TreeMap<V, Set<K>> valueKeyMap;
Review Comment:
` valueKeysMap`
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/PartitionBalancer.java:
##########
@@ -16,58 +16,237 @@
* specific language governing permissions and limitations
* under the License.
*/
+
package org.apache.iotdb.confignode.manager.load.balancer;
+import org.apache.iotdb.common.rpc.thrift.TConsensusGroupId;
+import org.apache.iotdb.common.rpc.thrift.TConsensusGroupType;
import org.apache.iotdb.common.rpc.thrift.TSeriesPartitionSlot;
+import org.apache.iotdb.common.rpc.thrift.TTimePartitionSlot;
import org.apache.iotdb.commons.partition.DataPartitionTable;
import org.apache.iotdb.commons.partition.SchemaPartitionTable;
+import org.apache.iotdb.commons.partition.SeriesPartitionTable;
+import org.apache.iotdb.commons.structure.BalanceTreeMap;
+import org.apache.iotdb.confignode.exception.DatabaseNotExistsException;
import org.apache.iotdb.confignode.exception.NoAvailableRegionGroupException;
import org.apache.iotdb.confignode.manager.IManager;
-import org.apache.iotdb.confignode.manager.load.balancer.partition.GreedyPartitionAllocator;
-import org.apache.iotdb.confignode.manager.load.balancer.partition.IPartitionAllocator;
+import org.apache.iotdb.confignode.manager.load.balancer.partition.DataPartitionPolicyTable;
+import org.apache.iotdb.confignode.manager.partition.PartitionManager;
+import org.apache.iotdb.confignode.manager.schema.ClusterSchemaManager;
import org.apache.iotdb.confignode.rpc.thrift.TTimeSlotList;
+import org.apache.iotdb.tsfile.utils.Pair;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import java.util.Comparator;
import java.util.List;
import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
/**
* The SeriesPartitionSlotBalancer provides interfaces to generate optimal Partition allocation and
* migration plans
*/
public class PartitionBalancer {
+ private static final Logger LOGGER = LoggerFactory.getLogger(PartitionBalancer.class);
+
private final IManager configManager;
+ // Map<DatabaseName, DataPartitionPolicyTable>
+ private final Map<String, DataPartitionPolicyTable> dataAllotTableMap;
+
public PartitionBalancer(IManager configManager) {
this.configManager = configManager;
+ this.dataAllotTableMap = new ConcurrentHashMap<>();
}
/**
* Allocate SchemaPartitions
*
* @param unassignedSchemaPartitionSlotsMap SchemaPartitionSlots that should be assigned
- * @return Map<StorageGroupName, SchemaPartitionTable>, the allocating result
+ * @return Map<DatabaseName, SchemaPartitionTable>, the allocating result
*/
public Map<String, SchemaPartitionTable> allocateSchemaPartition(
Map<String, List<TSeriesPartitionSlot>> unassignedSchemaPartitionSlotsMap)
throws NoAvailableRegionGroupException {
- return genPartitionAllocator().allocateSchemaPartition(unassignedSchemaPartitionSlotsMap);
+ Map<String, SchemaPartitionTable> result = new ConcurrentHashMap<>();
+
+ for (Map.Entry<String, List<TSeriesPartitionSlot>> slotsMapEntry :
+ unassignedSchemaPartitionSlotsMap.entrySet()) {
+ final String database = slotsMapEntry.getKey();
+ final List<TSeriesPartitionSlot> unassignedPartitionSlots = slotsMapEntry.getValue();
+
+ // Filter available SchemaRegionGroups and
+ // sort them by the number of allocated SchemaPartitions
+ BalanceTreeMap<TConsensusGroupId, Integer> counter = new BalanceTreeMap<>();
+ List<Pair<Long, TConsensusGroupId>> regionSlotsCounter =
+ getPartitionManager()
+ .getSortedRegionGroupSlotsCounter(database, TConsensusGroupType.SchemaRegion);
+ for (Pair<Long, TConsensusGroupId> pair : regionSlotsCounter) {
+ counter.put(pair.getRight(), pair.getLeft().intValue());
+ }
+
+ // Enumerate SeriesPartitionSlot
+ Map<TSeriesPartitionSlot, TConsensusGroupId> schemaPartitionMap = new ConcurrentHashMap<>();
+ for (TSeriesPartitionSlot seriesPartitionSlot : unassignedPartitionSlots) {
+ // Greedy allocation: allocate the unassigned SchemaPartition to
+ // the RegionGroup whose allocated SchemaPartitions is the least
+ TConsensusGroupId consensusGroupId = counter.getKeyWithMinValue();
+ schemaPartitionMap.put(seriesPartitionSlot, consensusGroupId);
+ counter.put(consensusGroupId, counter.get(consensusGroupId) + 1);
+ }
+ result.put(database, new SchemaPartitionTable(schemaPartitionMap));
+ }
+
+ return result;
}
/**
* Allocate DataPartitions
*
* @param unassignedDataPartitionSlotsMap DataPartitionSlots that should be assigned
- * @return Map<StorageGroupName, DataPartitionTable>, the allocating result
+ * @return Map<DatabaseName, DataPartitionTable>, the allocating result
*/
public Map<String, DataPartitionTable> allocateDataPartition(
Map<String, Map<TSeriesPartitionSlot, TTimeSlotList>> unassignedDataPartitionSlotsMap)
throws NoAvailableRegionGroupException {
- return genPartitionAllocator().allocateDataPartition(unassignedDataPartitionSlotsMap);
+ Map<String, DataPartitionTable> result = new ConcurrentHashMap<>();
+
+ for (Map.Entry<String, Map<TSeriesPartitionSlot, TTimeSlotList>> slotsMapEntry :
+ unassignedDataPartitionSlotsMap.entrySet()) {
+ final String database = slotsMapEntry.getKey();
+ final Map<TSeriesPartitionSlot, TTimeSlotList> unassignedPartitionSlotsMap =
+ slotsMapEntry.getValue();
+
+ // Filter available DataRegionGroups and
+ // sort them by the number of allocated DataPartitions
+ BalanceTreeMap<TConsensusGroupId, Integer> counter = new BalanceTreeMap<>();
+ List<Pair<Long, TConsensusGroupId>> regionSlotsCounter =
+ getPartitionManager()
+ .getSortedRegionGroupSlotsCounter(database, TConsensusGroupType.DataRegion);
+ for (Pair<Long, TConsensusGroupId> pair : regionSlotsCounter) {
+ counter.put(pair.getRight(), pair.getLeft().intValue());
+ }
+
+ DataPartitionPolicyTable allotTable = dataAllotTableMap.get(database);
+ allotTable.acquireLock();
+ DataPartitionTable dataPartitionTable = new DataPartitionTable();
+
+ try {
+ // Enumerate SeriesPartitionSlot
+ for (Map.Entry<TSeriesPartitionSlot, TTimeSlotList> seriesPartitionEntry :
+ unassignedPartitionSlotsMap.entrySet()) {
+ SeriesPartitionTable seriesPartitionTable = new SeriesPartitionTable();
+
+ // Enumerate TimePartitionSlot in ascending order
+ TSeriesPartitionSlot seriesPartitionSlot = seriesPartitionEntry.getKey();
+ List<TTimePartitionSlot> timePartitionSlots =
+ seriesPartitionEntry.getValue().getTimePartitionSlots();
+ timePartitionSlots.sort(Comparator.comparingLong(TTimePartitionSlot::getStartTime));
+
+ for (TTimePartitionSlot timePartitionSlot : timePartitionSlots) {
+
+ // 1. The historical DataPartition will try to inherit successor DataPartition first
+ TConsensusGroupId successor =
+ getPartitionManager()
+ .getSuccessorDataPartition(database, seriesPartitionSlot, timePartitionSlot);
+ if (successor != null && counter.containsKey(successor)) {
Review Comment:
Why we need `counter.containsKey(successor)`? Are only running RegionGroups allowed to exist inside them, and how is this guaranteed?
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/DeleteDatabaseProcedure.java:
##########
@@ -206,6 +206,7 @@ protected Flow executeFromState(ConfigNodeProcedureEnv env, DeleteStorageGroupSt
LOG.info(
"[DeleteDatabaseProcedure] Database: {} is deleted successfully",
deleteDatabaseSchema.getName());
+ env.getConfigManager().getLoadManager().clearPartitionBalancer();
Review Comment:
Whether only the corresponding database table should be clear?
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/partition/DataPartitionPolicyTable.java:
##########
@@ -0,0 +1,144 @@
+/*
+ * 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.iotdb.confignode.manager.load.balancer.partition;
+
+import org.apache.iotdb.common.rpc.thrift.TConsensusGroupId;
+import org.apache.iotdb.common.rpc.thrift.TSeriesPartitionSlot;
+import org.apache.iotdb.commons.structure.BalanceTreeMap;
+import org.apache.iotdb.confignode.conf.ConfigNodeConfig;
+import org.apache.iotdb.confignode.conf.ConfigNodeDescriptor;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.locks.ReentrantLock;
+
+public class DataPartitionPolicyTable {
+
+ private static final ConfigNodeConfig CONF = ConfigNodeDescriptor.getInstance().getConf();
+ private static final int SERIES_SLOT_NUM = CONF.getSeriesSlotNum();
+
+ private final ReentrantLock dataAllotTableLock;
+
+ // Map<SeriesPartitionSlot, RegionGroupId>
+ // The optimal allocation of SeriesSlots to RegionGroups
+ private final Map<TSeriesPartitionSlot, TConsensusGroupId> dataAllotMap;
+
+ // Map<RegionGroupId, SeriesPartitionSlot Count>
+ // The number of SeriesSlots allocated to each RegionGroup in dataAllotMap
+ private final BalanceTreeMap<TConsensusGroupId, Integer> seriesPartitionSlotCounter;
+
+ public DataPartitionPolicyTable() {
+ this.dataAllotTableLock = new ReentrantLock();
+ this.dataAllotMap = new HashMap<>();
+ this.seriesPartitionSlotCounter = new BalanceTreeMap<>();
+ }
+
+ /**
+ * Get or activate the specified SeriesPartitionSlot in dataAllotMap.
+ *
+ * @param seriesPartitionSlot The specified SeriesPartitionSlot
+ * @return The RegionGroupId of the specified SeriesPartitionSlot, activate when its empty yet
+ */
+ public TConsensusGroupId getRegionGroupIdOrActivateIfNecessary(
+ TSeriesPartitionSlot seriesPartitionSlot) {
+ if (dataAllotMap.containsKey(seriesPartitionSlot)) {
+ return dataAllotMap.get(seriesPartitionSlot);
+ }
+
+ TConsensusGroupId regionGroupId = seriesPartitionSlotCounter.getKeyWithMinValue();
+ dataAllotMap.put(seriesPartitionSlot, regionGroupId);
+ seriesPartitionSlotCounter.put(
+ regionGroupId, seriesPartitionSlotCounter.get(regionGroupId) + 1);
+ return regionGroupId;
+ }
+
+ /**
+ * Re-balance the allocation of SeriesSlots to RegionGroups.
+ *
+ * @param dataRegionGroups All DataRegionGroups currently in the Database
+ */
+ public void reBalanceDataPartitionPolicy(List<TConsensusGroupId> dataRegionGroups) {
+ dataAllotTableLock.lock();
+ try {
+ dataRegionGroups.forEach(
+ dataRegionGroup -> {
+ if (!seriesPartitionSlotCounter.containsKey(dataRegionGroup)) {
+ seriesPartitionSlotCounter.put(dataRegionGroup, 0);
+ }
+ });
+
+ // Enumerate all SeriesSlots randomly
+ List<TSeriesPartitionSlot> seriesPartitionSlots = new ArrayList<>();
+ for (int i = 0; i < SERIES_SLOT_NUM; i++) {
+ seriesPartitionSlots.add(new TSeriesPartitionSlot(i));
+ }
+ Collections.shuffle(seriesPartitionSlots);
+
+ int mu = SERIES_SLOT_NUM / dataRegionGroups.size();
+ for (TSeriesPartitionSlot seriesPartitionSlot : seriesPartitionSlots) {
+ if (!dataAllotMap.containsKey(seriesPartitionSlot)) {
+ continue;
+ }
+
+ TConsensusGroupId regionGroupId = dataAllotMap.get(seriesPartitionSlot);
+ int seriesPartitionSlotCount = seriesPartitionSlotCounter.get(regionGroupId);
+ if (seriesPartitionSlotCount > mu) {
+ // Remove from dataAllotMap if the number of SeriesSlots is greater than mu
+ dataAllotMap.remove(seriesPartitionSlot);
+ seriesPartitionSlotCounter.put(regionGroupId, seriesPartitionSlotCount - 1);
+ }
+ }
+ } finally {
+ dataAllotTableLock.unlock();
+ }
+ }
+
+ /** Only use this interface when init PartitionBalancer. */
+ public void setDataAllotMap(Map<TSeriesPartitionSlot, TConsensusGroupId> dataAllotMap) {
+ dataAllotTableLock.lock();
Review Comment:
move to the try scope
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/statemachine/CreateRegionGroupsProcedure.java:
##########
@@ -259,6 +275,7 @@ public void deserialize(ByteBuffer byteBuffer) {
ThriftCommonsSerDeUtils.deserializeTRegionReplicaSet(byteBuffer);
failedRegionReplicaSets.put(groupId, replica);
}
+ persistPlan.deserializeForProcedure(byteBuffer);
Review Comment:
For compatibility reasons, it is important to determine if there are still redundant bytes that can be read, and if so, deserialize them
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/PartitionBalancer.java:
##########
@@ -16,58 +16,237 @@
* specific language governing permissions and limitations
* under the License.
*/
+
package org.apache.iotdb.confignode.manager.load.balancer;
+import org.apache.iotdb.common.rpc.thrift.TConsensusGroupId;
+import org.apache.iotdb.common.rpc.thrift.TConsensusGroupType;
import org.apache.iotdb.common.rpc.thrift.TSeriesPartitionSlot;
+import org.apache.iotdb.common.rpc.thrift.TTimePartitionSlot;
import org.apache.iotdb.commons.partition.DataPartitionTable;
import org.apache.iotdb.commons.partition.SchemaPartitionTable;
+import org.apache.iotdb.commons.partition.SeriesPartitionTable;
+import org.apache.iotdb.commons.structure.BalanceTreeMap;
+import org.apache.iotdb.confignode.exception.DatabaseNotExistsException;
import org.apache.iotdb.confignode.exception.NoAvailableRegionGroupException;
import org.apache.iotdb.confignode.manager.IManager;
-import org.apache.iotdb.confignode.manager.load.balancer.partition.GreedyPartitionAllocator;
-import org.apache.iotdb.confignode.manager.load.balancer.partition.IPartitionAllocator;
+import org.apache.iotdb.confignode.manager.load.balancer.partition.DataPartitionPolicyTable;
+import org.apache.iotdb.confignode.manager.partition.PartitionManager;
+import org.apache.iotdb.confignode.manager.schema.ClusterSchemaManager;
import org.apache.iotdb.confignode.rpc.thrift.TTimeSlotList;
+import org.apache.iotdb.tsfile.utils.Pair;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import java.util.Comparator;
import java.util.List;
import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
/**
* The SeriesPartitionSlotBalancer provides interfaces to generate optimal Partition allocation and
* migration plans
*/
public class PartitionBalancer {
+ private static final Logger LOGGER = LoggerFactory.getLogger(PartitionBalancer.class);
+
private final IManager configManager;
+ // Map<DatabaseName, DataPartitionPolicyTable>
+ private final Map<String, DataPartitionPolicyTable> dataAllotTableMap;
+
public PartitionBalancer(IManager configManager) {
this.configManager = configManager;
+ this.dataAllotTableMap = new ConcurrentHashMap<>();
}
/**
* Allocate SchemaPartitions
*
* @param unassignedSchemaPartitionSlotsMap SchemaPartitionSlots that should be assigned
- * @return Map<StorageGroupName, SchemaPartitionTable>, the allocating result
+ * @return Map<DatabaseName, SchemaPartitionTable>, the allocating result
*/
public Map<String, SchemaPartitionTable> allocateSchemaPartition(
Map<String, List<TSeriesPartitionSlot>> unassignedSchemaPartitionSlotsMap)
throws NoAvailableRegionGroupException {
- return genPartitionAllocator().allocateSchemaPartition(unassignedSchemaPartitionSlotsMap);
+ Map<String, SchemaPartitionTable> result = new ConcurrentHashMap<>();
+
+ for (Map.Entry<String, List<TSeriesPartitionSlot>> slotsMapEntry :
+ unassignedSchemaPartitionSlotsMap.entrySet()) {
+ final String database = slotsMapEntry.getKey();
+ final List<TSeriesPartitionSlot> unassignedPartitionSlots = slotsMapEntry.getValue();
+
+ // Filter available SchemaRegionGroups and
+ // sort them by the number of allocated SchemaPartitions
+ BalanceTreeMap<TConsensusGroupId, Integer> counter = new BalanceTreeMap<>();
+ List<Pair<Long, TConsensusGroupId>> regionSlotsCounter =
+ getPartitionManager()
+ .getSortedRegionGroupSlotsCounter(database, TConsensusGroupType.SchemaRegion);
+ for (Pair<Long, TConsensusGroupId> pair : regionSlotsCounter) {
+ counter.put(pair.getRight(), pair.getLeft().intValue());
+ }
+
+ // Enumerate SeriesPartitionSlot
+ Map<TSeriesPartitionSlot, TConsensusGroupId> schemaPartitionMap = new ConcurrentHashMap<>();
+ for (TSeriesPartitionSlot seriesPartitionSlot : unassignedPartitionSlots) {
+ // Greedy allocation: allocate the unassigned SchemaPartition to
+ // the RegionGroup whose allocated SchemaPartitions is the least
+ TConsensusGroupId consensusGroupId = counter.getKeyWithMinValue();
+ schemaPartitionMap.put(seriesPartitionSlot, consensusGroupId);
+ counter.put(consensusGroupId, counter.get(consensusGroupId) + 1);
+ }
+ result.put(database, new SchemaPartitionTable(schemaPartitionMap));
+ }
+
+ return result;
}
/**
* Allocate DataPartitions
*
* @param unassignedDataPartitionSlotsMap DataPartitionSlots that should be assigned
- * @return Map<StorageGroupName, DataPartitionTable>, the allocating result
+ * @return Map<DatabaseName, DataPartitionTable>, the allocating result
*/
public Map<String, DataPartitionTable> allocateDataPartition(
Map<String, Map<TSeriesPartitionSlot, TTimeSlotList>> unassignedDataPartitionSlotsMap)
throws NoAvailableRegionGroupException {
- return genPartitionAllocator().allocateDataPartition(unassignedDataPartitionSlotsMap);
+ Map<String, DataPartitionTable> result = new ConcurrentHashMap<>();
+
+ for (Map.Entry<String, Map<TSeriesPartitionSlot, TTimeSlotList>> slotsMapEntry :
+ unassignedDataPartitionSlotsMap.entrySet()) {
+ final String database = slotsMapEntry.getKey();
+ final Map<TSeriesPartitionSlot, TTimeSlotList> unassignedPartitionSlotsMap =
+ slotsMapEntry.getValue();
+
+ // Filter available DataRegionGroups and
+ // sort them by the number of allocated DataPartitions
+ BalanceTreeMap<TConsensusGroupId, Integer> counter = new BalanceTreeMap<>();
+ List<Pair<Long, TConsensusGroupId>> regionSlotsCounter =
+ getPartitionManager()
+ .getSortedRegionGroupSlotsCounter(database, TConsensusGroupType.DataRegion);
+ for (Pair<Long, TConsensusGroupId> pair : regionSlotsCounter) {
+ counter.put(pair.getRight(), pair.getLeft().intValue());
+ }
+
+ DataPartitionPolicyTable allotTable = dataAllotTableMap.get(database);
+ allotTable.acquireLock();
+ DataPartitionTable dataPartitionTable = new DataPartitionTable();
+
+ try {
+ // Enumerate SeriesPartitionSlot
+ for (Map.Entry<TSeriesPartitionSlot, TTimeSlotList> seriesPartitionEntry :
+ unassignedPartitionSlotsMap.entrySet()) {
+ SeriesPartitionTable seriesPartitionTable = new SeriesPartitionTable();
+
+ // Enumerate TimePartitionSlot in ascending order
+ TSeriesPartitionSlot seriesPartitionSlot = seriesPartitionEntry.getKey();
+ List<TTimePartitionSlot> timePartitionSlots =
+ seriesPartitionEntry.getValue().getTimePartitionSlots();
+ timePartitionSlots.sort(Comparator.comparingLong(TTimePartitionSlot::getStartTime));
+
+ for (TTimePartitionSlot timePartitionSlot : timePartitionSlots) {
+
+ // 1. The historical DataPartition will try to inherit successor DataPartition first
+ TConsensusGroupId successor =
+ getPartitionManager()
+ .getSuccessorDataPartition(database, seriesPartitionSlot, timePartitionSlot);
+ if (successor != null && counter.containsKey(successor)) {
+ seriesPartitionTable.putDataPartition(timePartitionSlot, successor);
+ counter.put(successor, counter.get(successor) + 1);
+ continue;
+ }
+
+ // 2. Assign DataPartition base on the DataAllotTable
+ TConsensusGroupId allotGroupId =
+ allotTable.getRegionGroupIdOrActivateIfNecessary(seriesPartitionSlot);
+ if (counter.containsKey(allotGroupId)) {
+ seriesPartitionTable.putDataPartition(timePartitionSlot, allotGroupId);
+ counter.put(allotGroupId, counter.get(allotGroupId) + 1);
+ continue;
+ }
+
+ // 3. The allotDataRegionGroup is unavailable,
+ // try to inherit predecessor DataPartition
+ TConsensusGroupId predecessor =
+ getPartitionManager()
+ .getPredecessorDataPartition(database, seriesPartitionSlot, timePartitionSlot);
+ if (predecessor != null && counter.containsKey(predecessor)) {
+ seriesPartitionTable.putDataPartition(timePartitionSlot, predecessor);
+ counter.put(predecessor, counter.get(predecessor) + 1);
+ continue;
+ }
+
+ // 4. Assign the DataPartition to DataRegionGroup with the least DataPartitions
+ // If the above DataRegionGroups are unavailable
+ TConsensusGroupId greedyGroupId = counter.getKeyWithMinValue();
+ seriesPartitionTable.putDataPartition(timePartitionSlot, greedyGroupId);
+ counter.put(greedyGroupId, counter.get(greedyGroupId) + 1);
+ }
+
+ dataPartitionTable
+ .getDataPartitionMap()
+ .put(seriesPartitionEntry.getKey(), seriesPartitionTable);
+ }
+ } finally {
+ allotTable.releaseLock();
+ }
+ result.put(database, dataPartitionTable);
+ }
+
+ return result;
+ }
+
+ /**
+ * Re-balance the DataPartitionPolicyTable.
+ *
+ * @param database Database name
+ */
+ public void reBalanceDataPartitionPolicy(String database) {
+ try {
+ dataAllotTableMap
+ .computeIfAbsent(database, empty -> new DataPartitionPolicyTable())
+ .reBalanceDataPartitionPolicy(
+ getPartitionManager().getAllRegionGroupIds(database, TConsensusGroupType.DataRegion));
+ } catch (DatabaseNotExistsException e) {
+ LOGGER.error("Database {} not exists when updateDataAllotTable", database);
+ }
+ }
+
+ /** Set up the PartitionBalancer when the current ConfigNode becomes leader. */
+ public void setupPartitionBalancer() {
+ dataAllotTableMap.clear();
+ getClusterSchemaManager()
+ .getDatabaseNames()
+ .forEach(
+ database -> {
+ dataAllotTableMap.put(database, new DataPartitionPolicyTable());
Review Comment:
```
DataPartitionPolicyTable dataPartitionPolicyTable = new DataPartitionPolicyTable();
dataAllotTableMap.put(database, dataPartitionPolicyTable);
```
--
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
[GitHub] [iotdb] OneSizeFitsQuorum commented on a diff in pull request #10648: [IOTDB-6079] Cluster computing resource balance
Posted by "OneSizeFitsQuorum (via GitHub)" <gi...@apache.org>.
OneSizeFitsQuorum commented on code in PR #10648:
URL: https://github.com/apache/iotdb/pull/10648#discussion_r1279022193
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/PartitionBalancer.java:
##########
@@ -16,58 +16,237 @@
* specific language governing permissions and limitations
* under the License.
*/
+
package org.apache.iotdb.confignode.manager.load.balancer;
+import org.apache.iotdb.common.rpc.thrift.TConsensusGroupId;
+import org.apache.iotdb.common.rpc.thrift.TConsensusGroupType;
import org.apache.iotdb.common.rpc.thrift.TSeriesPartitionSlot;
+import org.apache.iotdb.common.rpc.thrift.TTimePartitionSlot;
import org.apache.iotdb.commons.partition.DataPartitionTable;
import org.apache.iotdb.commons.partition.SchemaPartitionTable;
+import org.apache.iotdb.commons.partition.SeriesPartitionTable;
+import org.apache.iotdb.commons.structure.BalanceTreeMap;
+import org.apache.iotdb.confignode.exception.DatabaseNotExistsException;
import org.apache.iotdb.confignode.exception.NoAvailableRegionGroupException;
import org.apache.iotdb.confignode.manager.IManager;
-import org.apache.iotdb.confignode.manager.load.balancer.partition.GreedyPartitionAllocator;
-import org.apache.iotdb.confignode.manager.load.balancer.partition.IPartitionAllocator;
+import org.apache.iotdb.confignode.manager.load.balancer.partition.DataPartitionPolicyTable;
+import org.apache.iotdb.confignode.manager.partition.PartitionManager;
+import org.apache.iotdb.confignode.manager.schema.ClusterSchemaManager;
import org.apache.iotdb.confignode.rpc.thrift.TTimeSlotList;
+import org.apache.iotdb.tsfile.utils.Pair;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import java.util.Comparator;
import java.util.List;
import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
/**
* The SeriesPartitionSlotBalancer provides interfaces to generate optimal Partition allocation and
* migration plans
*/
public class PartitionBalancer {
+ private static final Logger LOGGER = LoggerFactory.getLogger(PartitionBalancer.class);
+
private final IManager configManager;
+ // Map<DatabaseName, DataPartitionPolicyTable>
+ private final Map<String, DataPartitionPolicyTable> dataAllotTableMap;
+
public PartitionBalancer(IManager configManager) {
this.configManager = configManager;
+ this.dataAllotTableMap = new ConcurrentHashMap<>();
}
/**
* Allocate SchemaPartitions
*
* @param unassignedSchemaPartitionSlotsMap SchemaPartitionSlots that should be assigned
- * @return Map<StorageGroupName, SchemaPartitionTable>, the allocating result
+ * @return Map<DatabaseName, SchemaPartitionTable>, the allocating result
*/
public Map<String, SchemaPartitionTable> allocateSchemaPartition(
Map<String, List<TSeriesPartitionSlot>> unassignedSchemaPartitionSlotsMap)
throws NoAvailableRegionGroupException {
- return genPartitionAllocator().allocateSchemaPartition(unassignedSchemaPartitionSlotsMap);
+ Map<String, SchemaPartitionTable> result = new ConcurrentHashMap<>();
+
+ for (Map.Entry<String, List<TSeriesPartitionSlot>> slotsMapEntry :
+ unassignedSchemaPartitionSlotsMap.entrySet()) {
+ final String database = slotsMapEntry.getKey();
+ final List<TSeriesPartitionSlot> unassignedPartitionSlots = slotsMapEntry.getValue();
+
+ // Filter available SchemaRegionGroups and
+ // sort them by the number of allocated SchemaPartitions
+ BalanceTreeMap<TConsensusGroupId, Integer> counter = new BalanceTreeMap<>();
+ List<Pair<Long, TConsensusGroupId>> regionSlotsCounter =
+ getPartitionManager()
+ .getSortedRegionGroupSlotsCounter(database, TConsensusGroupType.SchemaRegion);
+ for (Pair<Long, TConsensusGroupId> pair : regionSlotsCounter) {
+ counter.put(pair.getRight(), pair.getLeft().intValue());
+ }
+
+ // Enumerate SeriesPartitionSlot
+ Map<TSeriesPartitionSlot, TConsensusGroupId> schemaPartitionMap = new ConcurrentHashMap<>();
+ for (TSeriesPartitionSlot seriesPartitionSlot : unassignedPartitionSlots) {
+ // Greedy allocation: allocate the unassigned SchemaPartition to
+ // the RegionGroup whose allocated SchemaPartitions is the least
+ TConsensusGroupId consensusGroupId = counter.getKeyWithMinValue();
+ schemaPartitionMap.put(seriesPartitionSlot, consensusGroupId);
+ counter.put(consensusGroupId, counter.get(consensusGroupId) + 1);
+ }
+ result.put(database, new SchemaPartitionTable(schemaPartitionMap));
+ }
+
+ return result;
}
/**
* Allocate DataPartitions
*
* @param unassignedDataPartitionSlotsMap DataPartitionSlots that should be assigned
- * @return Map<StorageGroupName, DataPartitionTable>, the allocating result
+ * @return Map<DatabaseName, DataPartitionTable>, the allocating result
*/
public Map<String, DataPartitionTable> allocateDataPartition(
Map<String, Map<TSeriesPartitionSlot, TTimeSlotList>> unassignedDataPartitionSlotsMap)
throws NoAvailableRegionGroupException {
- return genPartitionAllocator().allocateDataPartition(unassignedDataPartitionSlotsMap);
+ Map<String, DataPartitionTable> result = new ConcurrentHashMap<>();
+
+ for (Map.Entry<String, Map<TSeriesPartitionSlot, TTimeSlotList>> slotsMapEntry :
+ unassignedDataPartitionSlotsMap.entrySet()) {
+ final String database = slotsMapEntry.getKey();
+ final Map<TSeriesPartitionSlot, TTimeSlotList> unassignedPartitionSlotsMap =
+ slotsMapEntry.getValue();
+
+ // Filter available DataRegionGroups and
+ // sort them by the number of allocated DataPartitions
+ BalanceTreeMap<TConsensusGroupId, Integer> counter = new BalanceTreeMap<>();
+ List<Pair<Long, TConsensusGroupId>> regionSlotsCounter =
+ getPartitionManager()
+ .getSortedRegionGroupSlotsCounter(database, TConsensusGroupType.DataRegion);
+ for (Pair<Long, TConsensusGroupId> pair : regionSlotsCounter) {
+ counter.put(pair.getRight(), pair.getLeft().intValue());
+ }
+
+ DataPartitionPolicyTable allotTable = dataAllotTableMap.get(database);
+ allotTable.acquireLock();
Review Comment:
OK
--
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
[GitHub] [iotdb] CRZbulabula commented on a diff in pull request #10648: [IOTDB-6079] Cluster computing resource balance
Posted by "CRZbulabula (via GitHub)" <gi...@apache.org>.
CRZbulabula commented on code in PR #10648:
URL: https://github.com/apache/iotdb/pull/10648#discussion_r1278956322
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/partition/PartitionInfo.java:
##########
@@ -784,6 +822,19 @@ public Set<Integer> getAllSchemaPartition() {
return schemaPartitionSet;
}
+ /**
+ * Get the last DataAllotTable of the specified Database.
+ *
+ * @param database The specified Database
+ * @return The last DataAllotTable
+ */
+ public Map<TSeriesPartitionSlot, TConsensusGroupId> getLastDataAllotTable(String database) {
+ if (isDatabaseExisted(database)) {
+ return databasePartitionTables.get(database).getLastDataAllotTable();
+ }
+ return new HashMap<>();
Review Comment:
Fixed
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/PartitionBalancer.java:
##########
@@ -16,58 +16,237 @@
* specific language governing permissions and limitations
* under the License.
*/
+
package org.apache.iotdb.confignode.manager.load.balancer;
+import org.apache.iotdb.common.rpc.thrift.TConsensusGroupId;
+import org.apache.iotdb.common.rpc.thrift.TConsensusGroupType;
import org.apache.iotdb.common.rpc.thrift.TSeriesPartitionSlot;
+import org.apache.iotdb.common.rpc.thrift.TTimePartitionSlot;
import org.apache.iotdb.commons.partition.DataPartitionTable;
import org.apache.iotdb.commons.partition.SchemaPartitionTable;
+import org.apache.iotdb.commons.partition.SeriesPartitionTable;
+import org.apache.iotdb.commons.structure.BalanceTreeMap;
+import org.apache.iotdb.confignode.exception.DatabaseNotExistsException;
import org.apache.iotdb.confignode.exception.NoAvailableRegionGroupException;
import org.apache.iotdb.confignode.manager.IManager;
-import org.apache.iotdb.confignode.manager.load.balancer.partition.GreedyPartitionAllocator;
-import org.apache.iotdb.confignode.manager.load.balancer.partition.IPartitionAllocator;
+import org.apache.iotdb.confignode.manager.load.balancer.partition.DataPartitionPolicyTable;
+import org.apache.iotdb.confignode.manager.partition.PartitionManager;
+import org.apache.iotdb.confignode.manager.schema.ClusterSchemaManager;
import org.apache.iotdb.confignode.rpc.thrift.TTimeSlotList;
+import org.apache.iotdb.tsfile.utils.Pair;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import java.util.Comparator;
import java.util.List;
import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
/**
* The SeriesPartitionSlotBalancer provides interfaces to generate optimal Partition allocation and
* migration plans
*/
public class PartitionBalancer {
+ private static final Logger LOGGER = LoggerFactory.getLogger(PartitionBalancer.class);
+
private final IManager configManager;
+ // Map<DatabaseName, DataPartitionPolicyTable>
+ private final Map<String, DataPartitionPolicyTable> dataAllotTableMap;
+
public PartitionBalancer(IManager configManager) {
this.configManager = configManager;
+ this.dataAllotTableMap = new ConcurrentHashMap<>();
}
/**
* Allocate SchemaPartitions
*
* @param unassignedSchemaPartitionSlotsMap SchemaPartitionSlots that should be assigned
- * @return Map<StorageGroupName, SchemaPartitionTable>, the allocating result
+ * @return Map<DatabaseName, SchemaPartitionTable>, the allocating result
*/
public Map<String, SchemaPartitionTable> allocateSchemaPartition(
Map<String, List<TSeriesPartitionSlot>> unassignedSchemaPartitionSlotsMap)
throws NoAvailableRegionGroupException {
- return genPartitionAllocator().allocateSchemaPartition(unassignedSchemaPartitionSlotsMap);
+ Map<String, SchemaPartitionTable> result = new ConcurrentHashMap<>();
+
+ for (Map.Entry<String, List<TSeriesPartitionSlot>> slotsMapEntry :
+ unassignedSchemaPartitionSlotsMap.entrySet()) {
+ final String database = slotsMapEntry.getKey();
+ final List<TSeriesPartitionSlot> unassignedPartitionSlots = slotsMapEntry.getValue();
+
+ // Filter available SchemaRegionGroups and
+ // sort them by the number of allocated SchemaPartitions
+ BalanceTreeMap<TConsensusGroupId, Integer> counter = new BalanceTreeMap<>();
+ List<Pair<Long, TConsensusGroupId>> regionSlotsCounter =
+ getPartitionManager()
+ .getSortedRegionGroupSlotsCounter(database, TConsensusGroupType.SchemaRegion);
+ for (Pair<Long, TConsensusGroupId> pair : regionSlotsCounter) {
+ counter.put(pair.getRight(), pair.getLeft().intValue());
+ }
+
+ // Enumerate SeriesPartitionSlot
+ Map<TSeriesPartitionSlot, TConsensusGroupId> schemaPartitionMap = new ConcurrentHashMap<>();
+ for (TSeriesPartitionSlot seriesPartitionSlot : unassignedPartitionSlots) {
+ // Greedy allocation: allocate the unassigned SchemaPartition to
+ // the RegionGroup whose allocated SchemaPartitions is the least
+ TConsensusGroupId consensusGroupId = counter.getKeyWithMinValue();
+ schemaPartitionMap.put(seriesPartitionSlot, consensusGroupId);
+ counter.put(consensusGroupId, counter.get(consensusGroupId) + 1);
+ }
+ result.put(database, new SchemaPartitionTable(schemaPartitionMap));
+ }
+
+ return result;
}
/**
* Allocate DataPartitions
*
* @param unassignedDataPartitionSlotsMap DataPartitionSlots that should be assigned
- * @return Map<StorageGroupName, DataPartitionTable>, the allocating result
+ * @return Map<DatabaseName, DataPartitionTable>, the allocating result
*/
public Map<String, DataPartitionTable> allocateDataPartition(
Map<String, Map<TSeriesPartitionSlot, TTimeSlotList>> unassignedDataPartitionSlotsMap)
throws NoAvailableRegionGroupException {
- return genPartitionAllocator().allocateDataPartition(unassignedDataPartitionSlotsMap);
+ Map<String, DataPartitionTable> result = new ConcurrentHashMap<>();
Review Comment:
Fixed
--
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
[GitHub] [iotdb] sonarcloud[bot] commented on pull request #10648: Computing resource balancing
Posted by "sonarcloud[bot] (via GitHub)" <gi...@apache.org>.
sonarcloud[bot] commented on PR #10648:
URL: https://github.com/apache/iotdb/pull/10648#issuecomment-1646559792
SonarCloud Quality Gate failed. [](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=10648)
[](https://sonarcloud.io/project/issues?id=apache_iotdb&pullRequest=10648&resolved=false&types=BUG) [](https://sonarcloud.io/project/issues?id=apache_iotdb&pullRequest=10648&resolved=false&types=BUG) [1 Bug](https://sonarcloud.io/project/issues?id=apache_iotdb&pullRequest=10648&resolved=false&types=BUG)
[](https://sonarcloud.io/project/issues?id=apache_iotdb&pullRequest=10648&resolved=false&types=VULNERABILITY) [](https://sonarcloud.io/project/issues?id=apache_iotdb&pullRequest=10648&resolved=false&types=VULNERABILITY) [0 Vulnerabilities](https://sonarcloud.io/project/issues?id=apache_iotdb&pullRequest=10648&resolved=false&types=VULNERABILITY)
[](https://sonarcloud.io/project/security_hotspots?id=apache_iotdb&pullRequest=10648&resolved=false&types=SECURITY_HOTSPOT) [](https://sonarcloud.io/project/security_hotspots?id=apache_iotdb&pullRequest=10648&resolved=false&types=SECURITY_HOTSPOT) [1 Security Hotspot](https://sonarcloud.io/project/security_hotspots?id=apache_iotdb&pullRequest=10648&resolved=false&types=SECURITY_HOTSPOT)
[](https://sonarcloud.io/project/issues?id=apache_iotdb&pullRequest=10648&resolved=false&types=CODE_SMELL) [](https://sonarcloud.io/project/issues?id=apache_iotdb&pullRequest=10648&resolved=false&types=CODE_SMELL) [10 Code Smells](https://sonarcloud.io/project/issues?id=apache_iotdb&pullRequest=10648&resolved=false&types=CODE_SMELL)
[](https://sonarcloud.io/component_measures?id=apache_iotdb&pullRequest=10648&metric=new_coverage&view=list) [0.0% Coverage](https://sonarcloud.io/component_measures?id=apache_iotdb&pullRequest=10648&metric=new_coverage&view=list)
[](https://sonarcloud.io/component_measures?id=apache_iotdb&pullRequest=10648&metric=new_duplicated_lines_density&view=list) [0.0% Duplication](https://sonarcloud.io/component_measures?id=apache_iotdb&pullRequest=10648&metric=new_duplicated_lines_density&view=list)
 Catch issues before they fail your Quality Gate with our IDE extension  [SonarLint](https://www.sonarsource.com/products/sonarlint/features/connected-mode/?referrer=sonarcloud-welcome)
--
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
[GitHub] [iotdb] OneSizeFitsQuorum commented on a diff in pull request #10648: [IOTDB-6079] Cluster computing resource balance
Posted by "OneSizeFitsQuorum (via GitHub)" <gi...@apache.org>.
OneSizeFitsQuorum commented on code in PR #10648:
URL: https://github.com/apache/iotdb/pull/10648#discussion_r1279027325
##########
integration-test/src/test/java/org/apache/iotdb/session/it/IoTDBSessionSchemaTemplateIT.java:
##########
@@ -471,18 +471,22 @@ public void testHybridAutoExtendSchemaTemplate()
Assert.assertTrue(expectedSeries.isEmpty());
- deviceIds = Arrays.asList("root.db.v4.d1", "root.db.v4.d2");
- timestamps = Arrays.asList(1L, 1L);
measurements = Arrays.asList("a", "b", "c");
- allMeasurements = Arrays.asList(measurements, measurements);
- allTsDataTypes =
- Arrays.asList(
- Arrays.asList(TSDataType.FLOAT, TSDataType.FLOAT, TSDataType.TEXT),
- Arrays.asList(TSDataType.DOUBLE, TSDataType.DOUBLE, TSDataType.INT32));
- allValues = Arrays.asList(Arrays.asList(1f, 2f, "3"), Arrays.asList(1d, 2d, 3));
try {
- session.insertRecords(deviceIds, timestamps, allMeasurements, allTsDataTypes, allValues);
+ session.insertRecord(
Review Comment:
got it
--
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
[GitHub] [iotdb] OneSizeFitsQuorum commented on a diff in pull request #10648: [IOTDB-6079] Cluster computing resource balance
Posted by "OneSizeFitsQuorum (via GitHub)" <gi...@apache.org>.
OneSizeFitsQuorum commented on code in PR #10648:
URL: https://github.com/apache/iotdb/pull/10648#discussion_r1278772484
##########
integration-test/src/test/java/org/apache/iotdb/confignode/it/partition/IoTDBPartitionInheritPolicyIT.java:
##########
@@ -95,76 +91,173 @@ public static void tearDown() {
@Test
public void testDataPartitionInheritPolicy() throws Exception {
+ final long baseStartTime = 1000;
+ Map<TSeriesPartitionSlot, TConsensusGroupId> dataAllotTable1 = new ConcurrentHashMap<>();
- try (SyncConfigNodeIServiceClient client =
- (SyncConfigNodeIServiceClient) EnvFactory.getEnv().getLeaderConfigNodeConnection()) {
- TDataPartitionReq dataPartitionReq = new TDataPartitionReq();
- TDataPartitionTableResp dataPartitionTableResp;
- Map<String, Map<TSeriesPartitionSlot, TTimeSlotList>> partitionSlotsMap;
-
- for (int i = 0; i < storageGroupNum; i++) {
- String storageGroup = sg + i;
- for (int j = 0; j < testSeriesPartitionSlotNum; j += seriesPartitionBatchSize) {
- // Test inherit predecessor or successor
- boolean isAscending = (j / 10) % 2 == 0;
- int step = isAscending ? timePartitionBatchSize : -timePartitionBatchSize;
- int k = isAscending ? 0 : testTimePartitionSlotsNum - timePartitionBatchSize;
- while (0 <= k && k < testTimePartitionSlotsNum) {
- partitionSlotsMap =
- ConfigNodeTestUtils.constructPartitionSlotsMap(
- storageGroup,
- j,
- j + seriesPartitionBatchSize,
- k,
- k + timePartitionBatchSize,
- testTimePartitionInterval);
- // Let ConfigNode create DataPartition
- dataPartitionReq.setPartitionSlotsMap(partitionSlotsMap);
- for (int retry = 0; retry < 5; retry++) {
- // Build new Client since it's unstable
- try (SyncConfigNodeIServiceClient configNodeClient =
- (SyncConfigNodeIServiceClient)
- EnvFactory.getEnv().getLeaderConfigNodeConnection()) {
- dataPartitionTableResp =
- configNodeClient.getOrCreateDataPartitionTable(dataPartitionReq);
- if (dataPartitionTableResp != null
- && dataPartitionTableResp.getStatus().getCode()
- == TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
- ConfigNodeTestUtils.checkDataPartitionTable(
- storageGroup,
- j,
- j + seriesPartitionBatchSize,
- k,
- k + timePartitionBatchSize,
- testTimePartitionInterval,
- configNodeClient
- .getDataPartitionTable(dataPartitionReq)
- .getDataPartitionTable());
- break;
- }
- } catch (Exception e) {
- // Retry sometimes in order to avoid request timeout
- LOGGER.error(e.getMessage());
- TimeUnit.SECONDS.sleep(1);
+ // Test1: divide and inherit DataPartitions from scratch
+ for (long timePartitionSlot = baseStartTime;
+ timePartitionSlot < baseStartTime + testTimePartitionSlotsNum;
+ timePartitionSlot++) {
+ for (int seriesPartitionSlot = 0;
+ seriesPartitionSlot < testSeriesSlotNum;
+ seriesPartitionSlot += seriesPartitionSlotBatchSize) {
+ ConfigNodeTestUtils.getOrCreateDataPartitionWithRetry(
+ database,
+ seriesPartitionSlot,
+ seriesPartitionSlot + seriesPartitionSlotBatchSize,
+ timePartitionSlot,
+ timePartitionSlot + 1,
+ testTimePartitionInterval);
+ }
+ }
+
+ int mu = (int) (testSeriesSlotNum / testDataRegionPerDataNode);
+ TDataPartitionTableResp dataPartitionTableResp =
+ ConfigNodeTestUtils.getDataPartitionWithRetry(
+ database,
+ 0,
+ testSeriesSlotNum,
+ baseStartTime,
+ baseStartTime + testTimePartitionSlotsNum,
+ testTimePartitionInterval);
+ Assert.assertNotNull(dataPartitionTableResp);
+
+ // All DataRegionGroups divide all SeriesSlots evenly
+ final int expectedPartitionNum1 = mu * testTimePartitionSlotsNum;
+ Map<TConsensusGroupId, Integer> counter =
+ ConfigNodeTestUtils.countDataPartition(
+ dataPartitionTableResp.getDataPartitionTable().get(database));
+ counter.forEach((groupId, num) -> Assert.assertEquals(expectedPartitionNum1, num.intValue()));
+
+ // Test DataPartition inherit policy
+ dataPartitionTableResp
+ .getDataPartitionTable()
+ .get(database)
+ .forEach(
+ ((seriesPartitionSlot, timePartitionSlotMap) -> {
+ // All Timeslots belonging to the same SeriesSlot are allocated to the same
+ // DataRegionGroup
+ TConsensusGroupId groupId =
+ timePartitionSlotMap
+ .get(new TTimePartitionSlot(baseStartTime * testTimePartitionInterval))
+ .get(0);
+ timePartitionSlotMap.forEach(
+ (timePartitionSlot, groupIdList) ->
+ Assert.assertEquals(groupId, groupIdList.get(0)));
+ dataAllotTable1.put(seriesPartitionSlot, groupId);
+ }));
+
+ // Register a new DataNode to extend DataRegionGroups
+ EnvFactory.getEnv().registerNewDataNode(true);
+
+ // Test2: divide and inherit DataPartitions after extension
+ mu = (int) (testSeriesSlotNum / (testDataRegionPerDataNode * 2));
+ dataPartitionTableResp =
+ ConfigNodeTestUtils.getOrCreateDataPartitionWithRetry(
+ database,
+ 0,
+ testSeriesSlotNum,
+ baseStartTime + testTimePartitionSlotsNum,
+ baseStartTime + testTimePartitionSlotsNum + timePartitionBatchSize,
+ testTimePartitionInterval);
+ Assert.assertNotNull(dataPartitionTableResp);
+
+ // All DataRegionGroups divide all SeriesSlots evenly
+ counter =
+ ConfigNodeTestUtils.countDataPartition(
+ dataPartitionTableResp.getDataPartitionTable().get(database));
+ final int expectedPartitionNum2 = mu * timePartitionBatchSize;
+ counter.forEach((groupId, num) -> Assert.assertEquals(expectedPartitionNum2, num.intValue()));
+
+ // Test DataPartition inherit policy
+ Map<TSeriesPartitionSlot, TConsensusGroupId> dataAllotTable2 = new ConcurrentHashMap<>();
+ dataPartitionTableResp
+ .getDataPartitionTable()
+ .get(database)
+ .forEach(
+ ((seriesPartitionSlot, timePartitionSlotMap) -> {
+ // All Timeslots belonging to the same SeriesSlot are allocated to the same
+ // DataRegionGroup
+ TConsensusGroupId groupId =
+ timePartitionSlotMap
+ .get(
+ new TTimePartitionSlot(
+ (baseStartTime + testTimePartitionSlotsNum)
+ * testTimePartitionInterval))
+ .get(0);
+ timePartitionSlotMap.forEach(
+ (timePartitionSlot, groupIdList) ->
+ Assert.assertEquals(groupId, groupIdList.get(0)));
+
+ if (dataAllotTable1.containsValue(groupId)) {
+ // The DataRegionGroup has been inherited
+ Assert.assertTrue(dataAllotTable1.containsKey(seriesPartitionSlot));
+ Assert.assertEquals(dataAllotTable1.get(seriesPartitionSlot), groupId);
}
- }
- k += step;
- }
- }
+ dataAllotTable2.put(seriesPartitionSlot, groupId);
+ }));
+
+ // Test3: historical DataPartitions will inherit successor
+ Random random = new Random();
+ Set<Integer> allocatedSlots = new HashSet<>();
+ for (int i = 0; i < 10; i++) {
+ int slot = random.nextInt(testSeriesSlotNum);
+ while (allocatedSlots.contains(slot)) {
+ slot = random.nextInt(testSeriesSlotNum);
+ }
+ allocatedSlots.add(slot);
+ dataPartitionTableResp =
+ ConfigNodeTestUtils.getOrCreateDataPartitionWithRetry(
+ database,
+ slot,
+ slot + 1,
+ baseStartTime - 1,
+ baseStartTime,
+ testTimePartitionInterval);
+ Assert.assertNotNull(dataPartitionTableResp);
+
+ TSeriesPartitionSlot seriesPartitionSlot = new TSeriesPartitionSlot(slot);
+ TTimePartitionSlot timePartitionSlot =
+ new TTimePartitionSlot((baseStartTime - 1) * testTimePartitionInterval);
+ Assert.assertEquals(
+ dataAllotTable1.get(seriesPartitionSlot),
+ dataPartitionTableResp
+ .getDataPartitionTable()
+ .get(database)
+ .get(seriesPartitionSlot)
+ .get(timePartitionSlot)
+ .get(0));
+ }
+
+ // Test4: future DataPartitions will inherit predecessor
+ allocatedSlots.clear();
+ for (int i = 0; i < 10; i++) {
+ int slot = random.nextInt(testSeriesSlotNum);
+ while (allocatedSlots.contains(slot)) {
+ slot = random.nextInt(testSeriesSlotNum);
}
+ allocatedSlots.add(slot);
+ dataPartitionTableResp =
+ ConfigNodeTestUtils.getOrCreateDataPartitionWithRetry(
+ database,
+ slot,
+ slot + 1,
+ baseStartTime + 999,
Review Comment:
do not use magic number, you can use `testTimePartitionSlotsNum * 10 - 1` and `testTimePartitionSlotsNum * 10`
##########
integration-test/src/test/java/org/apache/iotdb/confignode/it/partition/IoTDBPartitionInheritPolicyIT.java:
##########
@@ -95,76 +91,173 @@ public static void tearDown() {
@Test
public void testDataPartitionInheritPolicy() throws Exception {
+ final long baseStartTime = 1000;
+ Map<TSeriesPartitionSlot, TConsensusGroupId> dataAllotTable1 = new ConcurrentHashMap<>();
- try (SyncConfigNodeIServiceClient client =
- (SyncConfigNodeIServiceClient) EnvFactory.getEnv().getLeaderConfigNodeConnection()) {
- TDataPartitionReq dataPartitionReq = new TDataPartitionReq();
- TDataPartitionTableResp dataPartitionTableResp;
- Map<String, Map<TSeriesPartitionSlot, TTimeSlotList>> partitionSlotsMap;
-
- for (int i = 0; i < storageGroupNum; i++) {
- String storageGroup = sg + i;
- for (int j = 0; j < testSeriesPartitionSlotNum; j += seriesPartitionBatchSize) {
- // Test inherit predecessor or successor
- boolean isAscending = (j / 10) % 2 == 0;
- int step = isAscending ? timePartitionBatchSize : -timePartitionBatchSize;
- int k = isAscending ? 0 : testTimePartitionSlotsNum - timePartitionBatchSize;
- while (0 <= k && k < testTimePartitionSlotsNum) {
- partitionSlotsMap =
- ConfigNodeTestUtils.constructPartitionSlotsMap(
- storageGroup,
- j,
- j + seriesPartitionBatchSize,
- k,
- k + timePartitionBatchSize,
- testTimePartitionInterval);
- // Let ConfigNode create DataPartition
- dataPartitionReq.setPartitionSlotsMap(partitionSlotsMap);
- for (int retry = 0; retry < 5; retry++) {
- // Build new Client since it's unstable
- try (SyncConfigNodeIServiceClient configNodeClient =
- (SyncConfigNodeIServiceClient)
- EnvFactory.getEnv().getLeaderConfigNodeConnection()) {
- dataPartitionTableResp =
- configNodeClient.getOrCreateDataPartitionTable(dataPartitionReq);
- if (dataPartitionTableResp != null
- && dataPartitionTableResp.getStatus().getCode()
- == TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
- ConfigNodeTestUtils.checkDataPartitionTable(
- storageGroup,
- j,
- j + seriesPartitionBatchSize,
- k,
- k + timePartitionBatchSize,
- testTimePartitionInterval,
- configNodeClient
- .getDataPartitionTable(dataPartitionReq)
- .getDataPartitionTable());
- break;
- }
- } catch (Exception e) {
- // Retry sometimes in order to avoid request timeout
- LOGGER.error(e.getMessage());
- TimeUnit.SECONDS.sleep(1);
+ // Test1: divide and inherit DataPartitions from scratch
+ for (long timePartitionSlot = baseStartTime;
+ timePartitionSlot < baseStartTime + testTimePartitionSlotsNum;
+ timePartitionSlot++) {
+ for (int seriesPartitionSlot = 0;
+ seriesPartitionSlot < testSeriesSlotNum;
+ seriesPartitionSlot += seriesPartitionSlotBatchSize) {
+ ConfigNodeTestUtils.getOrCreateDataPartitionWithRetry(
+ database,
+ seriesPartitionSlot,
+ seriesPartitionSlot + seriesPartitionSlotBatchSize,
+ timePartitionSlot,
+ timePartitionSlot + 1,
+ testTimePartitionInterval);
+ }
+ }
+
+ int mu = (int) (testSeriesSlotNum / testDataRegionPerDataNode);
+ TDataPartitionTableResp dataPartitionTableResp =
+ ConfigNodeTestUtils.getDataPartitionWithRetry(
+ database,
+ 0,
+ testSeriesSlotNum,
+ baseStartTime,
+ baseStartTime + testTimePartitionSlotsNum,
+ testTimePartitionInterval);
+ Assert.assertNotNull(dataPartitionTableResp);
+
+ // All DataRegionGroups divide all SeriesSlots evenly
+ final int expectedPartitionNum1 = mu * testTimePartitionSlotsNum;
+ Map<TConsensusGroupId, Integer> counter =
+ ConfigNodeTestUtils.countDataPartition(
+ dataPartitionTableResp.getDataPartitionTable().get(database));
+ counter.forEach((groupId, num) -> Assert.assertEquals(expectedPartitionNum1, num.intValue()));
+
+ // Test DataPartition inherit policy
+ dataPartitionTableResp
+ .getDataPartitionTable()
+ .get(database)
+ .forEach(
+ ((seriesPartitionSlot, timePartitionSlotMap) -> {
+ // All Timeslots belonging to the same SeriesSlot are allocated to the same
+ // DataRegionGroup
+ TConsensusGroupId groupId =
+ timePartitionSlotMap
+ .get(new TTimePartitionSlot(baseStartTime * testTimePartitionInterval))
+ .get(0);
+ timePartitionSlotMap.forEach(
+ (timePartitionSlot, groupIdList) ->
+ Assert.assertEquals(groupId, groupIdList.get(0)));
+ dataAllotTable1.put(seriesPartitionSlot, groupId);
+ }));
+
+ // Register a new DataNode to extend DataRegionGroups
+ EnvFactory.getEnv().registerNewDataNode(true);
+
+ // Test2: divide and inherit DataPartitions after extension
+ mu = (int) (testSeriesSlotNum / (testDataRegionPerDataNode * 2));
+ dataPartitionTableResp =
+ ConfigNodeTestUtils.getOrCreateDataPartitionWithRetry(
+ database,
+ 0,
+ testSeriesSlotNum,
+ baseStartTime + testTimePartitionSlotsNum,
+ baseStartTime + testTimePartitionSlotsNum + timePartitionBatchSize,
+ testTimePartitionInterval);
+ Assert.assertNotNull(dataPartitionTableResp);
+
+ // All DataRegionGroups divide all SeriesSlots evenly
+ counter =
+ ConfigNodeTestUtils.countDataPartition(
+ dataPartitionTableResp.getDataPartitionTable().get(database));
+ final int expectedPartitionNum2 = mu * timePartitionBatchSize;
+ counter.forEach((groupId, num) -> Assert.assertEquals(expectedPartitionNum2, num.intValue()));
+
+ // Test DataPartition inherit policy
+ Map<TSeriesPartitionSlot, TConsensusGroupId> dataAllotTable2 = new ConcurrentHashMap<>();
+ dataPartitionTableResp
+ .getDataPartitionTable()
+ .get(database)
+ .forEach(
+ ((seriesPartitionSlot, timePartitionSlotMap) -> {
+ // All Timeslots belonging to the same SeriesSlot are allocated to the same
+ // DataRegionGroup
+ TConsensusGroupId groupId =
+ timePartitionSlotMap
+ .get(
+ new TTimePartitionSlot(
+ (baseStartTime + testTimePartitionSlotsNum)
+ * testTimePartitionInterval))
+ .get(0);
+ timePartitionSlotMap.forEach(
+ (timePartitionSlot, groupIdList) ->
+ Assert.assertEquals(groupId, groupIdList.get(0)));
+
+ if (dataAllotTable1.containsValue(groupId)) {
+ // The DataRegionGroup has been inherited
+ Assert.assertTrue(dataAllotTable1.containsKey(seriesPartitionSlot));
+ Assert.assertEquals(dataAllotTable1.get(seriesPartitionSlot), groupId);
}
- }
- k += step;
- }
- }
+ dataAllotTable2.put(seriesPartitionSlot, groupId);
+ }));
+
+ // Test3: historical DataPartitions will inherit successor
+ Random random = new Random();
+ Set<Integer> allocatedSlots = new HashSet<>();
+ for (int i = 0; i < 10; i++) {
+ int slot = random.nextInt(testSeriesSlotNum);
+ while (allocatedSlots.contains(slot)) {
+ slot = random.nextInt(testSeriesSlotNum);
+ }
+ allocatedSlots.add(slot);
+ dataPartitionTableResp =
+ ConfigNodeTestUtils.getOrCreateDataPartitionWithRetry(
+ database,
+ slot,
+ slot + 1,
+ baseStartTime - 1,
+ baseStartTime,
+ testTimePartitionInterval);
+ Assert.assertNotNull(dataPartitionTableResp);
+
+ TSeriesPartitionSlot seriesPartitionSlot = new TSeriesPartitionSlot(slot);
+ TTimePartitionSlot timePartitionSlot =
+ new TTimePartitionSlot((baseStartTime - 1) * testTimePartitionInterval);
+ Assert.assertEquals(
+ dataAllotTable1.get(seriesPartitionSlot),
+ dataPartitionTableResp
+ .getDataPartitionTable()
+ .get(database)
+ .get(seriesPartitionSlot)
+ .get(timePartitionSlot)
+ .get(0));
+ }
+
+ // Test4: future DataPartitions will inherit predecessor
+ allocatedSlots.clear();
+ for (int i = 0; i < 10; i++) {
+ int slot = random.nextInt(testSeriesSlotNum);
+ while (allocatedSlots.contains(slot)) {
+ slot = random.nextInt(testSeriesSlotNum);
}
+ allocatedSlots.add(slot);
+ dataPartitionTableResp =
+ ConfigNodeTestUtils.getOrCreateDataPartitionWithRetry(
+ database,
+ slot,
+ slot + 1,
+ baseStartTime + 999,
+ baseStartTime + 1000,
+ testTimePartitionInterval);
+ Assert.assertNotNull(dataPartitionTableResp);
- // Test DataPartition inherit policy
- TShowRegionResp showRegionResp = client.showRegion(new TShowRegionReq());
- showRegionResp
- .getRegionInfoList()
- .forEach(
- regionInfo -> {
- // All Timeslots belonging to the same SeriesSlot are allocated to the same
- // DataRegionGroup
- Assert.assertEquals(
- regionInfo.getSeriesSlots() * testTimePartitionSlotsNum,
- regionInfo.getTimeSlots());
- });
+ TSeriesPartitionSlot seriesPartitionSlot = new TSeriesPartitionSlot(slot);
+ TTimePartitionSlot timePartitionSlot =
+ new TTimePartitionSlot((baseStartTime + 999) * testTimePartitionInterval);
Review Comment:
same
--
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
[GitHub] [iotdb] sonarcloud[bot] commented on pull request #10648: Computing resource balancing
Posted by "sonarcloud[bot] (via GitHub)" <gi...@apache.org>.
sonarcloud[bot] commented on PR #10648:
URL: https://github.com/apache/iotdb/pull/10648#issuecomment-1646587701
SonarCloud Quality Gate failed. [](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=10648)
[](https://sonarcloud.io/project/issues?id=apache_iotdb&pullRequest=10648&resolved=false&types=BUG) [](https://sonarcloud.io/project/issues?id=apache_iotdb&pullRequest=10648&resolved=false&types=BUG) [0 Bugs](https://sonarcloud.io/project/issues?id=apache_iotdb&pullRequest=10648&resolved=false&types=BUG)
[](https://sonarcloud.io/project/issues?id=apache_iotdb&pullRequest=10648&resolved=false&types=VULNERABILITY) [](https://sonarcloud.io/project/issues?id=apache_iotdb&pullRequest=10648&resolved=false&types=VULNERABILITY) [0 Vulnerabilities](https://sonarcloud.io/project/issues?id=apache_iotdb&pullRequest=10648&resolved=false&types=VULNERABILITY)
[](https://sonarcloud.io/project/security_hotspots?id=apache_iotdb&pullRequest=10648&resolved=false&types=SECURITY_HOTSPOT) [](https://sonarcloud.io/project/security_hotspots?id=apache_iotdb&pullRequest=10648&resolved=false&types=SECURITY_HOTSPOT) [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=apache_iotdb&pullRequest=10648&resolved=false&types=SECURITY_HOTSPOT)
[](https://sonarcloud.io/project/issues?id=apache_iotdb&pullRequest=10648&resolved=false&types=CODE_SMELL) [](https://sonarcloud.io/project/issues?id=apache_iotdb&pullRequest=10648&resolved=false&types=CODE_SMELL) [10 Code Smells](https://sonarcloud.io/project/issues?id=apache_iotdb&pullRequest=10648&resolved=false&types=CODE_SMELL)
[](https://sonarcloud.io/component_measures?id=apache_iotdb&pullRequest=10648&metric=new_coverage&view=list) [0.0% Coverage](https://sonarcloud.io/component_measures?id=apache_iotdb&pullRequest=10648&metric=new_coverage&view=list)
[](https://sonarcloud.io/component_measures?id=apache_iotdb&pullRequest=10648&metric=new_duplicated_lines_density&view=list) [0.0% Duplication](https://sonarcloud.io/component_measures?id=apache_iotdb&pullRequest=10648&metric=new_duplicated_lines_density&view=list)
 Catch issues before they fail your Quality Gate with our IDE extension  [SonarLint](https://www.sonarsource.com/products/sonarlint/features/connected-mode/?referrer=sonarcloud-welcome)
--
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
[GitHub] [iotdb] CRZbulabula commented on a diff in pull request #10648: [IOTDB-6079] Cluster computing resource balance
Posted by "CRZbulabula (via GitHub)" <gi...@apache.org>.
CRZbulabula commented on code in PR #10648:
URL: https://github.com/apache/iotdb/pull/10648#discussion_r1279235346
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/partition/DataPartitionPolicyTable.java:
##########
@@ -0,0 +1,144 @@
+/*
+ * 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.iotdb.confignode.manager.load.balancer.partition;
+
+import org.apache.iotdb.common.rpc.thrift.TConsensusGroupId;
+import org.apache.iotdb.common.rpc.thrift.TSeriesPartitionSlot;
+import org.apache.iotdb.commons.structure.BalanceTreeMap;
+import org.apache.iotdb.confignode.conf.ConfigNodeConfig;
+import org.apache.iotdb.confignode.conf.ConfigNodeDescriptor;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.locks.ReentrantLock;
+
+public class DataPartitionPolicyTable {
+
+ private static final ConfigNodeConfig CONF = ConfigNodeDescriptor.getInstance().getConf();
+ private static final int SERIES_SLOT_NUM = CONF.getSeriesSlotNum();
+
+ private final ReentrantLock dataAllotTableLock;
+
+ // Map<SeriesPartitionSlot, RegionGroupId>
+ // The optimal allocation of SeriesSlots to RegionGroups
+ private final Map<TSeriesPartitionSlot, TConsensusGroupId> dataAllotMap;
+
+ // Map<RegionGroupId, SeriesPartitionSlot Count>
+ // The number of SeriesSlots allocated to each RegionGroup in dataAllotMap
+ private final BalanceTreeMap<TConsensusGroupId, Integer> seriesPartitionSlotCounter;
+
+ public DataPartitionPolicyTable() {
+ this.dataAllotTableLock = new ReentrantLock();
+ this.dataAllotMap = new HashMap<>();
+ this.seriesPartitionSlotCounter = new BalanceTreeMap<>();
+ }
+
+ /**
+ * Get or activate the specified SeriesPartitionSlot in dataAllotMap.
+ *
+ * @param seriesPartitionSlot The specified SeriesPartitionSlot
+ * @return The RegionGroupId of the specified SeriesPartitionSlot, activate when its empty yet
+ */
+ public TConsensusGroupId getRegionGroupIdOrActivateIfNecessary(
+ TSeriesPartitionSlot seriesPartitionSlot) {
+ if (dataAllotMap.containsKey(seriesPartitionSlot)) {
+ return dataAllotMap.get(seriesPartitionSlot);
+ }
+
+ TConsensusGroupId regionGroupId = seriesPartitionSlotCounter.getKeyWithMinValue();
+ dataAllotMap.put(seriesPartitionSlot, regionGroupId);
+ seriesPartitionSlotCounter.put(
+ regionGroupId, seriesPartitionSlotCounter.get(regionGroupId) + 1);
+ return regionGroupId;
+ }
+
+ /**
+ * Re-balance the allocation of SeriesSlots to RegionGroups.
+ *
+ * @param dataRegionGroups All DataRegionGroups currently in the Database
+ */
+ public void reBalanceDataPartitionPolicy(List<TConsensusGroupId> dataRegionGroups) {
+ dataAllotTableLock.lock();
+ try {
+ dataRegionGroups.forEach(
+ dataRegionGroup -> {
+ if (!seriesPartitionSlotCounter.containsKey(dataRegionGroup)) {
+ seriesPartitionSlotCounter.put(dataRegionGroup, 0);
+ }
+ });
+
+ // Enumerate all SeriesSlots randomly
+ List<TSeriesPartitionSlot> seriesPartitionSlots = new ArrayList<>();
+ for (int i = 0; i < SERIES_SLOT_NUM; i++) {
+ seriesPartitionSlots.add(new TSeriesPartitionSlot(i));
+ }
+ Collections.shuffle(seriesPartitionSlots);
+
+ int mu = SERIES_SLOT_NUM / dataRegionGroups.size();
+ for (TSeriesPartitionSlot seriesPartitionSlot : seriesPartitionSlots) {
+ if (!dataAllotMap.containsKey(seriesPartitionSlot)) {
+ continue;
+ }
+
+ TConsensusGroupId regionGroupId = dataAllotMap.get(seriesPartitionSlot);
+ int seriesPartitionSlotCount = seriesPartitionSlotCounter.get(regionGroupId);
+ if (seriesPartitionSlotCount > mu) {
+ // Remove from dataAllotMap if the number of SeriesSlots is greater than mu
+ dataAllotMap.remove(seriesPartitionSlot);
+ seriesPartitionSlotCounter.put(regionGroupId, seriesPartitionSlotCount - 1);
+ }
+ }
+ } finally {
+ dataAllotTableLock.unlock();
+ }
+ }
+
+ /** Only use this interface when init PartitionBalancer. */
+ public void setDataAllotMap(Map<TSeriesPartitionSlot, TConsensusGroupId> dataAllotMap) {
+ dataAllotTableLock.lock();
Review Comment:
Fixed
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/statemachine/CreateRegionGroupsProcedure.java:
##########
@@ -259,6 +275,7 @@ public void deserialize(ByteBuffer byteBuffer) {
ThriftCommonsSerDeUtils.deserializeTRegionReplicaSet(byteBuffer);
failedRegionReplicaSets.put(groupId, replica);
}
+ persistPlan.deserializeForProcedure(byteBuffer);
Review Comment:
Fixed
--
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
[GitHub] [iotdb] CRZbulabula commented on a diff in pull request #10648: [IOTDB-6079] Cluster computing resource balance
Posted by "CRZbulabula (via GitHub)" <gi...@apache.org>.
CRZbulabula commented on code in PR #10648:
URL: https://github.com/apache/iotdb/pull/10648#discussion_r1278956716
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/DeleteDatabaseProcedure.java:
##########
@@ -206,6 +206,7 @@ protected Flow executeFromState(ConfigNodeProcedureEnv env, DeleteStorageGroupSt
LOG.info(
"[DeleteDatabaseProcedure] Database: {} is deleted successfully",
deleteDatabaseSchema.getName());
+ env.getConfigManager().getLoadManager().clearPartitionBalancer();
Review Comment:
My big mistake!!! Fixed now
--
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
[GitHub] [iotdb] sonarcloud[bot] commented on pull request #10648: [IOTDB-6079] Cluster computing resource balance
Posted by "sonarcloud[bot] (via GitHub)" <gi...@apache.org>.
sonarcloud[bot] commented on PR #10648:
URL: https://github.com/apache/iotdb/pull/10648#issuecomment-1656691628
SonarCloud Quality Gate failed. [](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=10648)
[](https://sonarcloud.io/project/issues?id=apache_iotdb&pullRequest=10648&resolved=false&types=BUG) [](https://sonarcloud.io/project/issues?id=apache_iotdb&pullRequest=10648&resolved=false&types=BUG) [0 Bugs](https://sonarcloud.io/project/issues?id=apache_iotdb&pullRequest=10648&resolved=false&types=BUG)
[](https://sonarcloud.io/project/issues?id=apache_iotdb&pullRequest=10648&resolved=false&types=VULNERABILITY) [](https://sonarcloud.io/project/issues?id=apache_iotdb&pullRequest=10648&resolved=false&types=VULNERABILITY) [0 Vulnerabilities](https://sonarcloud.io/project/issues?id=apache_iotdb&pullRequest=10648&resolved=false&types=VULNERABILITY)
[](https://sonarcloud.io/project/security_hotspots?id=apache_iotdb&pullRequest=10648&resolved=false&types=SECURITY_HOTSPOT) [](https://sonarcloud.io/project/security_hotspots?id=apache_iotdb&pullRequest=10648&resolved=false&types=SECURITY_HOTSPOT) [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=apache_iotdb&pullRequest=10648&resolved=false&types=SECURITY_HOTSPOT)
[](https://sonarcloud.io/project/issues?id=apache_iotdb&pullRequest=10648&resolved=false&types=CODE_SMELL) [](https://sonarcloud.io/project/issues?id=apache_iotdb&pullRequest=10648&resolved=false&types=CODE_SMELL) [7 Code Smells](https://sonarcloud.io/project/issues?id=apache_iotdb&pullRequest=10648&resolved=false&types=CODE_SMELL)
[](https://sonarcloud.io/component_measures?id=apache_iotdb&pullRequest=10648&metric=new_coverage&view=list) [0.0% Coverage](https://sonarcloud.io/component_measures?id=apache_iotdb&pullRequest=10648&metric=new_coverage&view=list)
[](https://sonarcloud.io/component_measures?id=apache_iotdb&pullRequest=10648&metric=new_duplicated_lines_density&view=list) [0.0% Duplication](https://sonarcloud.io/component_measures?id=apache_iotdb&pullRequest=10648&metric=new_duplicated_lines_density&view=list)
 Catch issues before they fail your Quality Gate with our IDE extension  [SonarLint](https://www.sonarsource.com/products/sonarlint/features/connected-mode/?referrer=sonarcloud-welcome)
--
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
[GitHub] [iotdb] sonarcloud[bot] commented on pull request #10648: Computing resource balancing
Posted by "sonarcloud[bot] (via GitHub)" <gi...@apache.org>.
sonarcloud[bot] commented on PR #10648:
URL: https://github.com/apache/iotdb/pull/10648#issuecomment-1645539263
SonarCloud Quality Gate failed. [](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=10648)
[](https://sonarcloud.io/project/issues?id=apache_iotdb&pullRequest=10648&resolved=false&types=BUG) [](https://sonarcloud.io/project/issues?id=apache_iotdb&pullRequest=10648&resolved=false&types=BUG) [0 Bugs](https://sonarcloud.io/project/issues?id=apache_iotdb&pullRequest=10648&resolved=false&types=BUG)
[](https://sonarcloud.io/project/issues?id=apache_iotdb&pullRequest=10648&resolved=false&types=VULNERABILITY) [](https://sonarcloud.io/project/issues?id=apache_iotdb&pullRequest=10648&resolved=false&types=VULNERABILITY) [0 Vulnerabilities](https://sonarcloud.io/project/issues?id=apache_iotdb&pullRequest=10648&resolved=false&types=VULNERABILITY)
[](https://sonarcloud.io/project/security_hotspots?id=apache_iotdb&pullRequest=10648&resolved=false&types=SECURITY_HOTSPOT) [](https://sonarcloud.io/project/security_hotspots?id=apache_iotdb&pullRequest=10648&resolved=false&types=SECURITY_HOTSPOT) [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=apache_iotdb&pullRequest=10648&resolved=false&types=SECURITY_HOTSPOT)
[](https://sonarcloud.io/project/issues?id=apache_iotdb&pullRequest=10648&resolved=false&types=CODE_SMELL) [](https://sonarcloud.io/project/issues?id=apache_iotdb&pullRequest=10648&resolved=false&types=CODE_SMELL) [17 Code Smells](https://sonarcloud.io/project/issues?id=apache_iotdb&pullRequest=10648&resolved=false&types=CODE_SMELL)
[](https://sonarcloud.io/component_measures?id=apache_iotdb&pullRequest=10648&metric=new_coverage&view=list) [0.0% Coverage](https://sonarcloud.io/component_measures?id=apache_iotdb&pullRequest=10648&metric=new_coverage&view=list)
[](https://sonarcloud.io/component_measures?id=apache_iotdb&pullRequest=10648&metric=new_duplicated_lines_density&view=list) [0.0% Duplication](https://sonarcloud.io/component_measures?id=apache_iotdb&pullRequest=10648&metric=new_duplicated_lines_density&view=list)
 Catch issues before they fail your Quality Gate with our IDE extension  [SonarLint](https://www.sonarsource.com/products/sonarlint/features/connected-mode/?referrer=sonarcloud-welcome)
--
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
[GitHub] [iotdb] OneSizeFitsQuorum commented on a diff in pull request #10648: [IOTDB-6079] Cluster computing resource balance
Posted by "OneSizeFitsQuorum (via GitHub)" <gi...@apache.org>.
OneSizeFitsQuorum commented on code in PR #10648:
URL: https://github.com/apache/iotdb/pull/10648#discussion_r1279061620
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/statemachine/CreateRegionGroupsProcedure.java:
##########
@@ -259,6 +275,7 @@ public void deserialize(ByteBuffer byteBuffer) {
ThriftCommonsSerDeUtils.deserializeTRegionReplicaSet(byteBuffer);
failedRegionReplicaSets.put(groupId, replica);
}
+ persistPlan.deserializeForProcedure(byteBuffer);
Review Comment:
The old confignode version without snapshot replays all Raftlogs on restart, where all procedure plans are called deserialize function. In this case, for the deserialization of newly added fields, It is a mistake not to determine whether there is margin. After deserializing, it will be replayed in configRegion, where the serialization method will be called to prevent NPE if it is not deserialized.
This is a screenshot of an error I accidentally made in this [pr](https://github.com/apache/iotdb/pull/10743/commits/384517502743cca22fc80dad89bba4b3e1cb61b8#diff-7b1570e04d8ec47ee0731a8670ea0879d67ee8cd96d2237521f46c8ed9c39570) while trying to change something like this
--
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
[GitHub] [iotdb] sonarcloud[bot] commented on pull request #10648: [IOTDB-6079] Cluster computing resource balance
Posted by "sonarcloud[bot] (via GitHub)" <gi...@apache.org>.
sonarcloud[bot] commented on PR #10648:
URL: https://github.com/apache/iotdb/pull/10648#issuecomment-1658328977
SonarCloud Quality Gate failed. [](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=10648)
[](https://sonarcloud.io/project/issues?id=apache_iotdb&pullRequest=10648&resolved=false&types=BUG) [](https://sonarcloud.io/project/issues?id=apache_iotdb&pullRequest=10648&resolved=false&types=BUG) [0 Bugs](https://sonarcloud.io/project/issues?id=apache_iotdb&pullRequest=10648&resolved=false&types=BUG)
[](https://sonarcloud.io/project/issues?id=apache_iotdb&pullRequest=10648&resolved=false&types=VULNERABILITY) [](https://sonarcloud.io/project/issues?id=apache_iotdb&pullRequest=10648&resolved=false&types=VULNERABILITY) [0 Vulnerabilities](https://sonarcloud.io/project/issues?id=apache_iotdb&pullRequest=10648&resolved=false&types=VULNERABILITY)
[](https://sonarcloud.io/project/security_hotspots?id=apache_iotdb&pullRequest=10648&resolved=false&types=SECURITY_HOTSPOT) [](https://sonarcloud.io/project/security_hotspots?id=apache_iotdb&pullRequest=10648&resolved=false&types=SECURITY_HOTSPOT) [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=apache_iotdb&pullRequest=10648&resolved=false&types=SECURITY_HOTSPOT)
[](https://sonarcloud.io/project/issues?id=apache_iotdb&pullRequest=10648&resolved=false&types=CODE_SMELL) [](https://sonarcloud.io/project/issues?id=apache_iotdb&pullRequest=10648&resolved=false&types=CODE_SMELL) [7 Code Smells](https://sonarcloud.io/project/issues?id=apache_iotdb&pullRequest=10648&resolved=false&types=CODE_SMELL)
[](https://sonarcloud.io/component_measures?id=apache_iotdb&pullRequest=10648&metric=new_coverage&view=list) [0.0% Coverage](https://sonarcloud.io/component_measures?id=apache_iotdb&pullRequest=10648&metric=new_coverage&view=list)
[](https://sonarcloud.io/component_measures?id=apache_iotdb&pullRequest=10648&metric=new_duplicated_lines_density&view=list) [0.0% Duplication](https://sonarcloud.io/component_measures?id=apache_iotdb&pullRequest=10648&metric=new_duplicated_lines_density&view=list)
 Catch issues before they fail your Quality Gate with our IDE extension  [SonarLint](https://www.sonarsource.com/products/sonarlint/features/connected-mode/?referrer=sonarcloud-welcome)
--
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
[GitHub] [iotdb] sonarcloud[bot] commented on pull request #10648: Computing resource balancing
Posted by "sonarcloud[bot] (via GitHub)" <gi...@apache.org>.
sonarcloud[bot] commented on PR #10648:
URL: https://github.com/apache/iotdb/pull/10648#issuecomment-1645676624
SonarCloud Quality Gate failed. [](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=10648)
[](https://sonarcloud.io/project/issues?id=apache_iotdb&pullRequest=10648&resolved=false&types=BUG) [](https://sonarcloud.io/project/issues?id=apache_iotdb&pullRequest=10648&resolved=false&types=BUG) [0 Bugs](https://sonarcloud.io/project/issues?id=apache_iotdb&pullRequest=10648&resolved=false&types=BUG)
[](https://sonarcloud.io/project/issues?id=apache_iotdb&pullRequest=10648&resolved=false&types=VULNERABILITY) [](https://sonarcloud.io/project/issues?id=apache_iotdb&pullRequest=10648&resolved=false&types=VULNERABILITY) [0 Vulnerabilities](https://sonarcloud.io/project/issues?id=apache_iotdb&pullRequest=10648&resolved=false&types=VULNERABILITY)
[](https://sonarcloud.io/project/security_hotspots?id=apache_iotdb&pullRequest=10648&resolved=false&types=SECURITY_HOTSPOT) [](https://sonarcloud.io/project/security_hotspots?id=apache_iotdb&pullRequest=10648&resolved=false&types=SECURITY_HOTSPOT) [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=apache_iotdb&pullRequest=10648&resolved=false&types=SECURITY_HOTSPOT)
[](https://sonarcloud.io/project/issues?id=apache_iotdb&pullRequest=10648&resolved=false&types=CODE_SMELL) [](https://sonarcloud.io/project/issues?id=apache_iotdb&pullRequest=10648&resolved=false&types=CODE_SMELL) [11 Code Smells](https://sonarcloud.io/project/issues?id=apache_iotdb&pullRequest=10648&resolved=false&types=CODE_SMELL)
[](https://sonarcloud.io/component_measures?id=apache_iotdb&pullRequest=10648&metric=new_coverage&view=list) [0.0% Coverage](https://sonarcloud.io/component_measures?id=apache_iotdb&pullRequest=10648&metric=new_coverage&view=list)
[](https://sonarcloud.io/component_measures?id=apache_iotdb&pullRequest=10648&metric=new_duplicated_lines_density&view=list) [0.0% Duplication](https://sonarcloud.io/component_measures?id=apache_iotdb&pullRequest=10648&metric=new_duplicated_lines_density&view=list)
 Catch issues before they fail your Quality Gate with our IDE extension  [SonarLint](https://www.sonarsource.com/products/sonarlint/features/connected-mode/?referrer=sonarcloud-welcome)
--
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
[GitHub] [iotdb] coveralls commented on pull request #10648: Computing resource balancing
Posted by "coveralls (via GitHub)" <gi...@apache.org>.
coveralls commented on PR #10648:
URL: https://github.com/apache/iotdb/pull/10648#issuecomment-1645736508
[](https://coveralls.io/builds/61499520)
coverage: 47.712% (+0.002%) from 47.71% when pulling **147a9eee96f8d17bae7ee7cc2e66e88544fae607 on Computing-resource-balancing** into **9090c5d35f7ce51c2c919ce37de77019272231ef on master**.
--
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
[GitHub] [iotdb] CRZbulabula commented on a diff in pull request #10648: [IOTDB-6079] Cluster computing resource balance
Posted by "CRZbulabula (via GitHub)" <gi...@apache.org>.
CRZbulabula commented on code in PR #10648:
URL: https://github.com/apache/iotdb/pull/10648#discussion_r1278956967
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/PartitionBalancer.java:
##########
@@ -16,58 +16,237 @@
* specific language governing permissions and limitations
* under the License.
*/
+
package org.apache.iotdb.confignode.manager.load.balancer;
+import org.apache.iotdb.common.rpc.thrift.TConsensusGroupId;
+import org.apache.iotdb.common.rpc.thrift.TConsensusGroupType;
import org.apache.iotdb.common.rpc.thrift.TSeriesPartitionSlot;
+import org.apache.iotdb.common.rpc.thrift.TTimePartitionSlot;
import org.apache.iotdb.commons.partition.DataPartitionTable;
import org.apache.iotdb.commons.partition.SchemaPartitionTable;
+import org.apache.iotdb.commons.partition.SeriesPartitionTable;
+import org.apache.iotdb.commons.structure.BalanceTreeMap;
+import org.apache.iotdb.confignode.exception.DatabaseNotExistsException;
import org.apache.iotdb.confignode.exception.NoAvailableRegionGroupException;
import org.apache.iotdb.confignode.manager.IManager;
-import org.apache.iotdb.confignode.manager.load.balancer.partition.GreedyPartitionAllocator;
-import org.apache.iotdb.confignode.manager.load.balancer.partition.IPartitionAllocator;
+import org.apache.iotdb.confignode.manager.load.balancer.partition.DataPartitionPolicyTable;
+import org.apache.iotdb.confignode.manager.partition.PartitionManager;
+import org.apache.iotdb.confignode.manager.schema.ClusterSchemaManager;
import org.apache.iotdb.confignode.rpc.thrift.TTimeSlotList;
+import org.apache.iotdb.tsfile.utils.Pair;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import java.util.Comparator;
import java.util.List;
import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
/**
* The SeriesPartitionSlotBalancer provides interfaces to generate optimal Partition allocation and
* migration plans
*/
public class PartitionBalancer {
+ private static final Logger LOGGER = LoggerFactory.getLogger(PartitionBalancer.class);
+
private final IManager configManager;
+ // Map<DatabaseName, DataPartitionPolicyTable>
+ private final Map<String, DataPartitionPolicyTable> dataAllotTableMap;
+
public PartitionBalancer(IManager configManager) {
this.configManager = configManager;
+ this.dataAllotTableMap = new ConcurrentHashMap<>();
}
/**
* Allocate SchemaPartitions
*
* @param unassignedSchemaPartitionSlotsMap SchemaPartitionSlots that should be assigned
- * @return Map<StorageGroupName, SchemaPartitionTable>, the allocating result
+ * @return Map<DatabaseName, SchemaPartitionTable>, the allocating result
*/
public Map<String, SchemaPartitionTable> allocateSchemaPartition(
Map<String, List<TSeriesPartitionSlot>> unassignedSchemaPartitionSlotsMap)
throws NoAvailableRegionGroupException {
- return genPartitionAllocator().allocateSchemaPartition(unassignedSchemaPartitionSlotsMap);
+ Map<String, SchemaPartitionTable> result = new ConcurrentHashMap<>();
+
+ for (Map.Entry<String, List<TSeriesPartitionSlot>> slotsMapEntry :
+ unassignedSchemaPartitionSlotsMap.entrySet()) {
+ final String database = slotsMapEntry.getKey();
+ final List<TSeriesPartitionSlot> unassignedPartitionSlots = slotsMapEntry.getValue();
+
+ // Filter available SchemaRegionGroups and
+ // sort them by the number of allocated SchemaPartitions
+ BalanceTreeMap<TConsensusGroupId, Integer> counter = new BalanceTreeMap<>();
+ List<Pair<Long, TConsensusGroupId>> regionSlotsCounter =
+ getPartitionManager()
+ .getSortedRegionGroupSlotsCounter(database, TConsensusGroupType.SchemaRegion);
+ for (Pair<Long, TConsensusGroupId> pair : regionSlotsCounter) {
+ counter.put(pair.getRight(), pair.getLeft().intValue());
+ }
+
+ // Enumerate SeriesPartitionSlot
+ Map<TSeriesPartitionSlot, TConsensusGroupId> schemaPartitionMap = new ConcurrentHashMap<>();
+ for (TSeriesPartitionSlot seriesPartitionSlot : unassignedPartitionSlots) {
+ // Greedy allocation: allocate the unassigned SchemaPartition to
+ // the RegionGroup whose allocated SchemaPartitions is the least
+ TConsensusGroupId consensusGroupId = counter.getKeyWithMinValue();
+ schemaPartitionMap.put(seriesPartitionSlot, consensusGroupId);
+ counter.put(consensusGroupId, counter.get(consensusGroupId) + 1);
+ }
+ result.put(database, new SchemaPartitionTable(schemaPartitionMap));
+ }
+
+ return result;
}
/**
* Allocate DataPartitions
*
* @param unassignedDataPartitionSlotsMap DataPartitionSlots that should be assigned
- * @return Map<StorageGroupName, DataPartitionTable>, the allocating result
+ * @return Map<DatabaseName, DataPartitionTable>, the allocating result
*/
public Map<String, DataPartitionTable> allocateDataPartition(
Map<String, Map<TSeriesPartitionSlot, TTimeSlotList>> unassignedDataPartitionSlotsMap)
throws NoAvailableRegionGroupException {
- return genPartitionAllocator().allocateDataPartition(unassignedDataPartitionSlotsMap);
+ Map<String, DataPartitionTable> result = new ConcurrentHashMap<>();
+
+ for (Map.Entry<String, Map<TSeriesPartitionSlot, TTimeSlotList>> slotsMapEntry :
+ unassignedDataPartitionSlotsMap.entrySet()) {
+ final String database = slotsMapEntry.getKey();
+ final Map<TSeriesPartitionSlot, TTimeSlotList> unassignedPartitionSlotsMap =
+ slotsMapEntry.getValue();
+
+ // Filter available DataRegionGroups and
+ // sort them by the number of allocated DataPartitions
+ BalanceTreeMap<TConsensusGroupId, Integer> counter = new BalanceTreeMap<>();
+ List<Pair<Long, TConsensusGroupId>> regionSlotsCounter =
+ getPartitionManager()
+ .getSortedRegionGroupSlotsCounter(database, TConsensusGroupType.DataRegion);
+ for (Pair<Long, TConsensusGroupId> pair : regionSlotsCounter) {
+ counter.put(pair.getRight(), pair.getLeft().intValue());
+ }
+
+ DataPartitionPolicyTable allotTable = dataAllotTableMap.get(database);
+ allotTable.acquireLock();
+ DataPartitionTable dataPartitionTable = new DataPartitionTable();
+
+ try {
+ // Enumerate SeriesPartitionSlot
+ for (Map.Entry<TSeriesPartitionSlot, TTimeSlotList> seriesPartitionEntry :
+ unassignedPartitionSlotsMap.entrySet()) {
+ SeriesPartitionTable seriesPartitionTable = new SeriesPartitionTable();
+
+ // Enumerate TimePartitionSlot in ascending order
+ TSeriesPartitionSlot seriesPartitionSlot = seriesPartitionEntry.getKey();
+ List<TTimePartitionSlot> timePartitionSlots =
+ seriesPartitionEntry.getValue().getTimePartitionSlots();
+ timePartitionSlots.sort(Comparator.comparingLong(TTimePartitionSlot::getStartTime));
+
+ for (TTimePartitionSlot timePartitionSlot : timePartitionSlots) {
+
+ // 1. The historical DataPartition will try to inherit successor DataPartition first
+ TConsensusGroupId successor =
+ getPartitionManager()
+ .getSuccessorDataPartition(database, seriesPartitionSlot, timePartitionSlot);
+ if (successor != null && counter.containsKey(successor)) {
+ seriesPartitionTable.putDataPartition(timePartitionSlot, successor);
+ counter.put(successor, counter.get(successor) + 1);
+ continue;
+ }
+
+ // 2. Assign DataPartition base on the DataAllotTable
+ TConsensusGroupId allotGroupId =
+ allotTable.getRegionGroupIdOrActivateIfNecessary(seriesPartitionSlot);
+ if (counter.containsKey(allotGroupId)) {
+ seriesPartitionTable.putDataPartition(timePartitionSlot, allotGroupId);
+ counter.put(allotGroupId, counter.get(allotGroupId) + 1);
+ continue;
+ }
+
+ // 3. The allotDataRegionGroup is unavailable,
+ // try to inherit predecessor DataPartition
+ TConsensusGroupId predecessor =
+ getPartitionManager()
+ .getPredecessorDataPartition(database, seriesPartitionSlot, timePartitionSlot);
+ if (predecessor != null && counter.containsKey(predecessor)) {
+ seriesPartitionTable.putDataPartition(timePartitionSlot, predecessor);
+ counter.put(predecessor, counter.get(predecessor) + 1);
+ continue;
+ }
+
+ // 4. Assign the DataPartition to DataRegionGroup with the least DataPartitions
+ // If the above DataRegionGroups are unavailable
+ TConsensusGroupId greedyGroupId = counter.getKeyWithMinValue();
+ seriesPartitionTable.putDataPartition(timePartitionSlot, greedyGroupId);
+ counter.put(greedyGroupId, counter.get(greedyGroupId) + 1);
+ }
+
+ dataPartitionTable
+ .getDataPartitionMap()
+ .put(seriesPartitionEntry.getKey(), seriesPartitionTable);
+ }
+ } finally {
+ allotTable.releaseLock();
+ }
+ result.put(database, dataPartitionTable);
+ }
+
+ return result;
+ }
+
+ /**
+ * Re-balance the DataPartitionPolicyTable.
+ *
+ * @param database Database name
+ */
+ public void reBalanceDataPartitionPolicy(String database) {
+ try {
+ dataAllotTableMap
+ .computeIfAbsent(database, empty -> new DataPartitionPolicyTable())
+ .reBalanceDataPartitionPolicy(
+ getPartitionManager().getAllRegionGroupIds(database, TConsensusGroupType.DataRegion));
+ } catch (DatabaseNotExistsException e) {
+ LOGGER.error("Database {} not exists when updateDataAllotTable", database);
+ }
+ }
+
+ /** Set up the PartitionBalancer when the current ConfigNode becomes leader. */
+ public void setupPartitionBalancer() {
+ dataAllotTableMap.clear();
+ getClusterSchemaManager()
+ .getDatabaseNames()
+ .forEach(
+ database -> {
+ dataAllotTableMap.put(database, new DataPartitionPolicyTable());
Review Comment:
Fixed
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/PartitionBalancer.java:
##########
@@ -16,58 +16,237 @@
* specific language governing permissions and limitations
* under the License.
*/
+
package org.apache.iotdb.confignode.manager.load.balancer;
+import org.apache.iotdb.common.rpc.thrift.TConsensusGroupId;
+import org.apache.iotdb.common.rpc.thrift.TConsensusGroupType;
import org.apache.iotdb.common.rpc.thrift.TSeriesPartitionSlot;
+import org.apache.iotdb.common.rpc.thrift.TTimePartitionSlot;
import org.apache.iotdb.commons.partition.DataPartitionTable;
import org.apache.iotdb.commons.partition.SchemaPartitionTable;
+import org.apache.iotdb.commons.partition.SeriesPartitionTable;
+import org.apache.iotdb.commons.structure.BalanceTreeMap;
+import org.apache.iotdb.confignode.exception.DatabaseNotExistsException;
import org.apache.iotdb.confignode.exception.NoAvailableRegionGroupException;
import org.apache.iotdb.confignode.manager.IManager;
-import org.apache.iotdb.confignode.manager.load.balancer.partition.GreedyPartitionAllocator;
-import org.apache.iotdb.confignode.manager.load.balancer.partition.IPartitionAllocator;
+import org.apache.iotdb.confignode.manager.load.balancer.partition.DataPartitionPolicyTable;
+import org.apache.iotdb.confignode.manager.partition.PartitionManager;
+import org.apache.iotdb.confignode.manager.schema.ClusterSchemaManager;
import org.apache.iotdb.confignode.rpc.thrift.TTimeSlotList;
+import org.apache.iotdb.tsfile.utils.Pair;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import java.util.Comparator;
import java.util.List;
import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
/**
* The SeriesPartitionSlotBalancer provides interfaces to generate optimal Partition allocation and
* migration plans
*/
public class PartitionBalancer {
+ private static final Logger LOGGER = LoggerFactory.getLogger(PartitionBalancer.class);
+
private final IManager configManager;
+ // Map<DatabaseName, DataPartitionPolicyTable>
+ private final Map<String, DataPartitionPolicyTable> dataAllotTableMap;
+
public PartitionBalancer(IManager configManager) {
this.configManager = configManager;
+ this.dataAllotTableMap = new ConcurrentHashMap<>();
}
/**
* Allocate SchemaPartitions
*
* @param unassignedSchemaPartitionSlotsMap SchemaPartitionSlots that should be assigned
- * @return Map<StorageGroupName, SchemaPartitionTable>, the allocating result
+ * @return Map<DatabaseName, SchemaPartitionTable>, the allocating result
*/
public Map<String, SchemaPartitionTable> allocateSchemaPartition(
Map<String, List<TSeriesPartitionSlot>> unassignedSchemaPartitionSlotsMap)
throws NoAvailableRegionGroupException {
- return genPartitionAllocator().allocateSchemaPartition(unassignedSchemaPartitionSlotsMap);
+ Map<String, SchemaPartitionTable> result = new ConcurrentHashMap<>();
+
+ for (Map.Entry<String, List<TSeriesPartitionSlot>> slotsMapEntry :
+ unassignedSchemaPartitionSlotsMap.entrySet()) {
+ final String database = slotsMapEntry.getKey();
+ final List<TSeriesPartitionSlot> unassignedPartitionSlots = slotsMapEntry.getValue();
+
+ // Filter available SchemaRegionGroups and
+ // sort them by the number of allocated SchemaPartitions
+ BalanceTreeMap<TConsensusGroupId, Integer> counter = new BalanceTreeMap<>();
+ List<Pair<Long, TConsensusGroupId>> regionSlotsCounter =
+ getPartitionManager()
+ .getSortedRegionGroupSlotsCounter(database, TConsensusGroupType.SchemaRegion);
+ for (Pair<Long, TConsensusGroupId> pair : regionSlotsCounter) {
+ counter.put(pair.getRight(), pair.getLeft().intValue());
+ }
+
+ // Enumerate SeriesPartitionSlot
+ Map<TSeriesPartitionSlot, TConsensusGroupId> schemaPartitionMap = new ConcurrentHashMap<>();
Review Comment:
Fixed
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/PartitionBalancer.java:
##########
@@ -16,58 +16,237 @@
* specific language governing permissions and limitations
* under the License.
*/
+
package org.apache.iotdb.confignode.manager.load.balancer;
+import org.apache.iotdb.common.rpc.thrift.TConsensusGroupId;
+import org.apache.iotdb.common.rpc.thrift.TConsensusGroupType;
import org.apache.iotdb.common.rpc.thrift.TSeriesPartitionSlot;
+import org.apache.iotdb.common.rpc.thrift.TTimePartitionSlot;
import org.apache.iotdb.commons.partition.DataPartitionTable;
import org.apache.iotdb.commons.partition.SchemaPartitionTable;
+import org.apache.iotdb.commons.partition.SeriesPartitionTable;
+import org.apache.iotdb.commons.structure.BalanceTreeMap;
+import org.apache.iotdb.confignode.exception.DatabaseNotExistsException;
import org.apache.iotdb.confignode.exception.NoAvailableRegionGroupException;
import org.apache.iotdb.confignode.manager.IManager;
-import org.apache.iotdb.confignode.manager.load.balancer.partition.GreedyPartitionAllocator;
-import org.apache.iotdb.confignode.manager.load.balancer.partition.IPartitionAllocator;
+import org.apache.iotdb.confignode.manager.load.balancer.partition.DataPartitionPolicyTable;
+import org.apache.iotdb.confignode.manager.partition.PartitionManager;
+import org.apache.iotdb.confignode.manager.schema.ClusterSchemaManager;
import org.apache.iotdb.confignode.rpc.thrift.TTimeSlotList;
+import org.apache.iotdb.tsfile.utils.Pair;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import java.util.Comparator;
import java.util.List;
import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
/**
* The SeriesPartitionSlotBalancer provides interfaces to generate optimal Partition allocation and
* migration plans
*/
public class PartitionBalancer {
+ private static final Logger LOGGER = LoggerFactory.getLogger(PartitionBalancer.class);
+
private final IManager configManager;
+ // Map<DatabaseName, DataPartitionPolicyTable>
+ private final Map<String, DataPartitionPolicyTable> dataAllotTableMap;
+
public PartitionBalancer(IManager configManager) {
this.configManager = configManager;
+ this.dataAllotTableMap = new ConcurrentHashMap<>();
}
/**
* Allocate SchemaPartitions
*
* @param unassignedSchemaPartitionSlotsMap SchemaPartitionSlots that should be assigned
- * @return Map<StorageGroupName, SchemaPartitionTable>, the allocating result
+ * @return Map<DatabaseName, SchemaPartitionTable>, the allocating result
*/
public Map<String, SchemaPartitionTable> allocateSchemaPartition(
Map<String, List<TSeriesPartitionSlot>> unassignedSchemaPartitionSlotsMap)
throws NoAvailableRegionGroupException {
- return genPartitionAllocator().allocateSchemaPartition(unassignedSchemaPartitionSlotsMap);
+ Map<String, SchemaPartitionTable> result = new ConcurrentHashMap<>();
Review Comment:
Fixed
--
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
[GitHub] [iotdb] CRZbulabula commented on a diff in pull request #10648: [IOTDB-6079] Cluster computing resource balance
Posted by "CRZbulabula (via GitHub)" <gi...@apache.org>.
CRZbulabula commented on code in PR #10648:
URL: https://github.com/apache/iotdb/pull/10648#discussion_r1278957916
##########
iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/structure/BalanceTreeMap.java:
##########
@@ -0,0 +1,98 @@
+/*
+ * 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.iotdb.commons.structure;
+
+import org.apache.iotdb.commons.utils.TestOnly;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Set;
+import java.util.TreeMap;
+
+public class BalanceTreeMap<K, V extends Comparable<V>> {
+
+ private final HashMap<K, V> keyValueMap;
+ private final TreeMap<V, Set<K>> valueKeyMap;
Review Comment:
Fixed
--
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
[GitHub] [iotdb] CRZbulabula commented on a diff in pull request #10648: [IOTDB-6079] Cluster computing resource balance
Posted by "CRZbulabula (via GitHub)" <gi...@apache.org>.
CRZbulabula commented on code in PR #10648:
URL: https://github.com/apache/iotdb/pull/10648#discussion_r1278957300
##########
iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/structure/BalanceTreeMap.java:
##########
@@ -0,0 +1,98 @@
+/*
+ * 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.iotdb.commons.structure;
+
+import org.apache.iotdb.commons.utils.TestOnly;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Set;
+import java.util.TreeMap;
+
+public class BalanceTreeMap<K, V extends Comparable<V>> {
Review Comment:
Added
##########
iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/structure/BalanceTreeMap.java:
##########
@@ -0,0 +1,98 @@
+/*
+ * 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.iotdb.commons.structure;
+
+import org.apache.iotdb.commons.utils.TestOnly;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Set;
+import java.util.TreeMap;
+
+public class BalanceTreeMap<K, V extends Comparable<V>> {
+
+ private final HashMap<K, V> keyValueMap;
+ private final TreeMap<V, Set<K>> valueKeyMap;
+
+ public BalanceTreeMap() {
+ this.keyValueMap = new HashMap<>();
+ this.valueKeyMap = new TreeMap<>();
+ }
+
+ /**
+ * Put or modify a key-value pair.
+ *
+ * @param key Key
+ * @param value Value
+ */
+ public void put(K key, V value) {
+ V oldValue = keyValueMap.getOrDefault(key, null);
+
+ // Update keyValueMap
+ keyValueMap.put(key, value);
+
+ // Update valueKeyMap
+ if (oldValue != null) {
+ valueKeyMap.get(oldValue).remove(key);
+ if (valueKeyMap.get(oldValue).isEmpty()) {
+ valueKeyMap.remove(oldValue);
+ }
+ }
+ valueKeyMap.computeIfAbsent(value, empty -> new HashSet<>()).add(key);
+ }
+
+ /**
+ * Get key with minimum value.
+ *
+ * @return Key with minimum value
+ */
+ public K getKeyWithMinValue() {
+ return valueKeyMap.firstEntry().getValue().iterator().next();
+ }
+
+ public V get(K key) {
+ return keyValueMap.getOrDefault(key, null);
+ }
+
+ public boolean containsKey(K key) {
+ return keyValueMap.containsKey(key);
+ }
+
+ public int size() {
+ return keyValueMap.size();
+ }
+
+ @TestOnly
+ public void remove(K key) {
Review Comment:
Fixed
##########
iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/structure/BalanceTreeMap.java:
##########
@@ -0,0 +1,98 @@
+/*
+ * 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.iotdb.commons.structure;
+
+import org.apache.iotdb.commons.utils.TestOnly;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Set;
+import java.util.TreeMap;
+
+public class BalanceTreeMap<K, V extends Comparable<V>> {
+
+ private final HashMap<K, V> keyValueMap;
+ private final TreeMap<V, Set<K>> valueKeyMap;
+
+ public BalanceTreeMap() {
+ this.keyValueMap = new HashMap<>();
+ this.valueKeyMap = new TreeMap<>();
+ }
+
+ /**
+ * Put or modify a key-value pair.
+ *
+ * @param key Key
+ * @param value Value
+ */
+ public void put(K key, V value) {
+ V oldValue = keyValueMap.getOrDefault(key, null);
+
+ // Update keyValueMap
+ keyValueMap.put(key, value);
+
+ // Update valueKeyMap
+ if (oldValue != null) {
Review Comment:
Fixed
##########
iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/structure/BalanceTreeMap.java:
##########
@@ -0,0 +1,98 @@
+/*
+ * 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.iotdb.commons.structure;
+
+import org.apache.iotdb.commons.utils.TestOnly;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Set;
+import java.util.TreeMap;
+
+public class BalanceTreeMap<K, V extends Comparable<V>> {
+
+ private final HashMap<K, V> keyValueMap;
+ private final TreeMap<V, Set<K>> valueKeyMap;
+
+ public BalanceTreeMap() {
+ this.keyValueMap = new HashMap<>();
+ this.valueKeyMap = new TreeMap<>();
+ }
+
+ /**
+ * Put or modify a key-value pair.
+ *
+ * @param key Key
+ * @param value Value
+ */
+ public void put(K key, V value) {
+ V oldValue = keyValueMap.getOrDefault(key, null);
+
+ // Update keyValueMap
+ keyValueMap.put(key, value);
Review Comment:
Fixed
--
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
[GitHub] [iotdb] CRZbulabula commented on a diff in pull request #10648: [IOTDB-6079] Cluster computing resource balance
Posted by "CRZbulabula (via GitHub)" <gi...@apache.org>.
CRZbulabula commented on code in PR #10648:
URL: https://github.com/apache/iotdb/pull/10648#discussion_r1278955878
##########
integration-test/src/test/java/org/apache/iotdb/confignode/it/partition/IoTDBPartitionInheritPolicyIT.java:
##########
@@ -95,76 +91,173 @@ public static void tearDown() {
@Test
public void testDataPartitionInheritPolicy() throws Exception {
+ final long baseStartTime = 1000;
+ Map<TSeriesPartitionSlot, TConsensusGroupId> dataAllotTable1 = new ConcurrentHashMap<>();
- try (SyncConfigNodeIServiceClient client =
- (SyncConfigNodeIServiceClient) EnvFactory.getEnv().getLeaderConfigNodeConnection()) {
- TDataPartitionReq dataPartitionReq = new TDataPartitionReq();
- TDataPartitionTableResp dataPartitionTableResp;
- Map<String, Map<TSeriesPartitionSlot, TTimeSlotList>> partitionSlotsMap;
-
- for (int i = 0; i < storageGroupNum; i++) {
- String storageGroup = sg + i;
- for (int j = 0; j < testSeriesPartitionSlotNum; j += seriesPartitionBatchSize) {
- // Test inherit predecessor or successor
- boolean isAscending = (j / 10) % 2 == 0;
- int step = isAscending ? timePartitionBatchSize : -timePartitionBatchSize;
- int k = isAscending ? 0 : testTimePartitionSlotsNum - timePartitionBatchSize;
- while (0 <= k && k < testTimePartitionSlotsNum) {
- partitionSlotsMap =
- ConfigNodeTestUtils.constructPartitionSlotsMap(
- storageGroup,
- j,
- j + seriesPartitionBatchSize,
- k,
- k + timePartitionBatchSize,
- testTimePartitionInterval);
- // Let ConfigNode create DataPartition
- dataPartitionReq.setPartitionSlotsMap(partitionSlotsMap);
- for (int retry = 0; retry < 5; retry++) {
- // Build new Client since it's unstable
- try (SyncConfigNodeIServiceClient configNodeClient =
- (SyncConfigNodeIServiceClient)
- EnvFactory.getEnv().getLeaderConfigNodeConnection()) {
- dataPartitionTableResp =
- configNodeClient.getOrCreateDataPartitionTable(dataPartitionReq);
- if (dataPartitionTableResp != null
- && dataPartitionTableResp.getStatus().getCode()
- == TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
- ConfigNodeTestUtils.checkDataPartitionTable(
- storageGroup,
- j,
- j + seriesPartitionBatchSize,
- k,
- k + timePartitionBatchSize,
- testTimePartitionInterval,
- configNodeClient
- .getDataPartitionTable(dataPartitionReq)
- .getDataPartitionTable());
- break;
- }
- } catch (Exception e) {
- // Retry sometimes in order to avoid request timeout
- LOGGER.error(e.getMessage());
- TimeUnit.SECONDS.sleep(1);
+ // Test1: divide and inherit DataPartitions from scratch
+ for (long timePartitionSlot = baseStartTime;
+ timePartitionSlot < baseStartTime + testTimePartitionSlotsNum;
+ timePartitionSlot++) {
+ for (int seriesPartitionSlot = 0;
+ seriesPartitionSlot < testSeriesSlotNum;
+ seriesPartitionSlot += seriesPartitionSlotBatchSize) {
+ ConfigNodeTestUtils.getOrCreateDataPartitionWithRetry(
+ database,
+ seriesPartitionSlot,
+ seriesPartitionSlot + seriesPartitionSlotBatchSize,
+ timePartitionSlot,
+ timePartitionSlot + 1,
+ testTimePartitionInterval);
+ }
+ }
+
+ int mu = (int) (testSeriesSlotNum / testDataRegionPerDataNode);
+ TDataPartitionTableResp dataPartitionTableResp =
+ ConfigNodeTestUtils.getDataPartitionWithRetry(
+ database,
+ 0,
+ testSeriesSlotNum,
+ baseStartTime,
+ baseStartTime + testTimePartitionSlotsNum,
+ testTimePartitionInterval);
+ Assert.assertNotNull(dataPartitionTableResp);
+
+ // All DataRegionGroups divide all SeriesSlots evenly
+ final int expectedPartitionNum1 = mu * testTimePartitionSlotsNum;
+ Map<TConsensusGroupId, Integer> counter =
+ ConfigNodeTestUtils.countDataPartition(
+ dataPartitionTableResp.getDataPartitionTable().get(database));
+ counter.forEach((groupId, num) -> Assert.assertEquals(expectedPartitionNum1, num.intValue()));
+
+ // Test DataPartition inherit policy
+ dataPartitionTableResp
+ .getDataPartitionTable()
+ .get(database)
+ .forEach(
+ ((seriesPartitionSlot, timePartitionSlotMap) -> {
+ // All Timeslots belonging to the same SeriesSlot are allocated to the same
+ // DataRegionGroup
+ TConsensusGroupId groupId =
+ timePartitionSlotMap
+ .get(new TTimePartitionSlot(baseStartTime * testTimePartitionInterval))
+ .get(0);
+ timePartitionSlotMap.forEach(
+ (timePartitionSlot, groupIdList) ->
+ Assert.assertEquals(groupId, groupIdList.get(0)));
+ dataAllotTable1.put(seriesPartitionSlot, groupId);
+ }));
+
+ // Register a new DataNode to extend DataRegionGroups
+ EnvFactory.getEnv().registerNewDataNode(true);
+
+ // Test2: divide and inherit DataPartitions after extension
+ mu = (int) (testSeriesSlotNum / (testDataRegionPerDataNode * 2));
+ dataPartitionTableResp =
+ ConfigNodeTestUtils.getOrCreateDataPartitionWithRetry(
+ database,
+ 0,
+ testSeriesSlotNum,
+ baseStartTime + testTimePartitionSlotsNum,
+ baseStartTime + testTimePartitionSlotsNum + timePartitionBatchSize,
+ testTimePartitionInterval);
+ Assert.assertNotNull(dataPartitionTableResp);
+
+ // All DataRegionGroups divide all SeriesSlots evenly
+ counter =
+ ConfigNodeTestUtils.countDataPartition(
+ dataPartitionTableResp.getDataPartitionTable().get(database));
+ final int expectedPartitionNum2 = mu * timePartitionBatchSize;
+ counter.forEach((groupId, num) -> Assert.assertEquals(expectedPartitionNum2, num.intValue()));
+
+ // Test DataPartition inherit policy
+ Map<TSeriesPartitionSlot, TConsensusGroupId> dataAllotTable2 = new ConcurrentHashMap<>();
+ dataPartitionTableResp
+ .getDataPartitionTable()
+ .get(database)
+ .forEach(
+ ((seriesPartitionSlot, timePartitionSlotMap) -> {
+ // All Timeslots belonging to the same SeriesSlot are allocated to the same
+ // DataRegionGroup
+ TConsensusGroupId groupId =
+ timePartitionSlotMap
+ .get(
+ new TTimePartitionSlot(
+ (baseStartTime + testTimePartitionSlotsNum)
+ * testTimePartitionInterval))
+ .get(0);
+ timePartitionSlotMap.forEach(
+ (timePartitionSlot, groupIdList) ->
+ Assert.assertEquals(groupId, groupIdList.get(0)));
+
+ if (dataAllotTable1.containsValue(groupId)) {
+ // The DataRegionGroup has been inherited
Review Comment:
Added
--
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
[GitHub] [iotdb] sonarcloud[bot] commented on pull request #10648: [IOTDB-6079] Cluster computing resource balance
Posted by "sonarcloud[bot] (via GitHub)" <gi...@apache.org>.
sonarcloud[bot] commented on PR #10648:
URL: https://github.com/apache/iotdb/pull/10648#issuecomment-1656580620
SonarCloud Quality Gate failed. [](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=10648)
[](https://sonarcloud.io/project/issues?id=apache_iotdb&pullRequest=10648&resolved=false&types=BUG) [](https://sonarcloud.io/project/issues?id=apache_iotdb&pullRequest=10648&resolved=false&types=BUG) [0 Bugs](https://sonarcloud.io/project/issues?id=apache_iotdb&pullRequest=10648&resolved=false&types=BUG)
[](https://sonarcloud.io/project/issues?id=apache_iotdb&pullRequest=10648&resolved=false&types=VULNERABILITY) [](https://sonarcloud.io/project/issues?id=apache_iotdb&pullRequest=10648&resolved=false&types=VULNERABILITY) [0 Vulnerabilities](https://sonarcloud.io/project/issues?id=apache_iotdb&pullRequest=10648&resolved=false&types=VULNERABILITY)
[](https://sonarcloud.io/project/security_hotspots?id=apache_iotdb&pullRequest=10648&resolved=false&types=SECURITY_HOTSPOT) [](https://sonarcloud.io/project/security_hotspots?id=apache_iotdb&pullRequest=10648&resolved=false&types=SECURITY_HOTSPOT) [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=apache_iotdb&pullRequest=10648&resolved=false&types=SECURITY_HOTSPOT)
[](https://sonarcloud.io/project/issues?id=apache_iotdb&pullRequest=10648&resolved=false&types=CODE_SMELL) [](https://sonarcloud.io/project/issues?id=apache_iotdb&pullRequest=10648&resolved=false&types=CODE_SMELL) [7 Code Smells](https://sonarcloud.io/project/issues?id=apache_iotdb&pullRequest=10648&resolved=false&types=CODE_SMELL)
[](https://sonarcloud.io/component_measures?id=apache_iotdb&pullRequest=10648&metric=new_coverage&view=list) [0.0% Coverage](https://sonarcloud.io/component_measures?id=apache_iotdb&pullRequest=10648&metric=new_coverage&view=list)
[](https://sonarcloud.io/component_measures?id=apache_iotdb&pullRequest=10648&metric=new_duplicated_lines_density&view=list) [0.0% Duplication](https://sonarcloud.io/component_measures?id=apache_iotdb&pullRequest=10648&metric=new_duplicated_lines_density&view=list)
 Catch issues before they fail your Quality Gate with our IDE extension  [SonarLint](https://www.sonarsource.com/products/sonarlint/features/connected-mode/?referrer=sonarcloud-welcome)
--
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
[GitHub] [iotdb] CRZbulabula commented on a diff in pull request #10648: [IOTDB-6079] Cluster computing resource balance
Posted by "CRZbulabula (via GitHub)" <gi...@apache.org>.
CRZbulabula commented on code in PR #10648:
URL: https://github.com/apache/iotdb/pull/10648#discussion_r1278943781
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/PartitionBalancer.java:
##########
@@ -16,58 +16,237 @@
* specific language governing permissions and limitations
* under the License.
*/
+
package org.apache.iotdb.confignode.manager.load.balancer;
+import org.apache.iotdb.common.rpc.thrift.TConsensusGroupId;
+import org.apache.iotdb.common.rpc.thrift.TConsensusGroupType;
import org.apache.iotdb.common.rpc.thrift.TSeriesPartitionSlot;
+import org.apache.iotdb.common.rpc.thrift.TTimePartitionSlot;
import org.apache.iotdb.commons.partition.DataPartitionTable;
import org.apache.iotdb.commons.partition.SchemaPartitionTable;
+import org.apache.iotdb.commons.partition.SeriesPartitionTable;
+import org.apache.iotdb.commons.structure.BalanceTreeMap;
+import org.apache.iotdb.confignode.exception.DatabaseNotExistsException;
import org.apache.iotdb.confignode.exception.NoAvailableRegionGroupException;
import org.apache.iotdb.confignode.manager.IManager;
-import org.apache.iotdb.confignode.manager.load.balancer.partition.GreedyPartitionAllocator;
-import org.apache.iotdb.confignode.manager.load.balancer.partition.IPartitionAllocator;
+import org.apache.iotdb.confignode.manager.load.balancer.partition.DataPartitionPolicyTable;
+import org.apache.iotdb.confignode.manager.partition.PartitionManager;
+import org.apache.iotdb.confignode.manager.schema.ClusterSchemaManager;
import org.apache.iotdb.confignode.rpc.thrift.TTimeSlotList;
+import org.apache.iotdb.tsfile.utils.Pair;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import java.util.Comparator;
import java.util.List;
import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
/**
* The SeriesPartitionSlotBalancer provides interfaces to generate optimal Partition allocation and
* migration plans
*/
public class PartitionBalancer {
+ private static final Logger LOGGER = LoggerFactory.getLogger(PartitionBalancer.class);
+
private final IManager configManager;
+ // Map<DatabaseName, DataPartitionPolicyTable>
+ private final Map<String, DataPartitionPolicyTable> dataAllotTableMap;
+
public PartitionBalancer(IManager configManager) {
this.configManager = configManager;
+ this.dataAllotTableMap = new ConcurrentHashMap<>();
}
/**
* Allocate SchemaPartitions
*
* @param unassignedSchemaPartitionSlotsMap SchemaPartitionSlots that should be assigned
- * @return Map<StorageGroupName, SchemaPartitionTable>, the allocating result
+ * @return Map<DatabaseName, SchemaPartitionTable>, the allocating result
*/
public Map<String, SchemaPartitionTable> allocateSchemaPartition(
Map<String, List<TSeriesPartitionSlot>> unassignedSchemaPartitionSlotsMap)
throws NoAvailableRegionGroupException {
- return genPartitionAllocator().allocateSchemaPartition(unassignedSchemaPartitionSlotsMap);
+ Map<String, SchemaPartitionTable> result = new ConcurrentHashMap<>();
+
+ for (Map.Entry<String, List<TSeriesPartitionSlot>> slotsMapEntry :
+ unassignedSchemaPartitionSlotsMap.entrySet()) {
+ final String database = slotsMapEntry.getKey();
+ final List<TSeriesPartitionSlot> unassignedPartitionSlots = slotsMapEntry.getValue();
+
+ // Filter available SchemaRegionGroups and
+ // sort them by the number of allocated SchemaPartitions
+ BalanceTreeMap<TConsensusGroupId, Integer> counter = new BalanceTreeMap<>();
+ List<Pair<Long, TConsensusGroupId>> regionSlotsCounter =
+ getPartitionManager()
+ .getSortedRegionGroupSlotsCounter(database, TConsensusGroupType.SchemaRegion);
+ for (Pair<Long, TConsensusGroupId> pair : regionSlotsCounter) {
+ counter.put(pair.getRight(), pair.getLeft().intValue());
+ }
+
+ // Enumerate SeriesPartitionSlot
+ Map<TSeriesPartitionSlot, TConsensusGroupId> schemaPartitionMap = new ConcurrentHashMap<>();
+ for (TSeriesPartitionSlot seriesPartitionSlot : unassignedPartitionSlots) {
+ // Greedy allocation: allocate the unassigned SchemaPartition to
+ // the RegionGroup whose allocated SchemaPartitions is the least
+ TConsensusGroupId consensusGroupId = counter.getKeyWithMinValue();
+ schemaPartitionMap.put(seriesPartitionSlot, consensusGroupId);
+ counter.put(consensusGroupId, counter.get(consensusGroupId) + 1);
+ }
+ result.put(database, new SchemaPartitionTable(schemaPartitionMap));
+ }
+
+ return result;
}
/**
* Allocate DataPartitions
*
* @param unassignedDataPartitionSlotsMap DataPartitionSlots that should be assigned
- * @return Map<StorageGroupName, DataPartitionTable>, the allocating result
+ * @return Map<DatabaseName, DataPartitionTable>, the allocating result
*/
public Map<String, DataPartitionTable> allocateDataPartition(
Map<String, Map<TSeriesPartitionSlot, TTimeSlotList>> unassignedDataPartitionSlotsMap)
throws NoAvailableRegionGroupException {
- return genPartitionAllocator().allocateDataPartition(unassignedDataPartitionSlotsMap);
+ Map<String, DataPartitionTable> result = new ConcurrentHashMap<>();
+
+ for (Map.Entry<String, Map<TSeriesPartitionSlot, TTimeSlotList>> slotsMapEntry :
+ unassignedDataPartitionSlotsMap.entrySet()) {
+ final String database = slotsMapEntry.getKey();
+ final Map<TSeriesPartitionSlot, TTimeSlotList> unassignedPartitionSlotsMap =
+ slotsMapEntry.getValue();
+
+ // Filter available DataRegionGroups and
+ // sort them by the number of allocated DataPartitions
+ BalanceTreeMap<TConsensusGroupId, Integer> counter = new BalanceTreeMap<>();
+ List<Pair<Long, TConsensusGroupId>> regionSlotsCounter =
+ getPartitionManager()
+ .getSortedRegionGroupSlotsCounter(database, TConsensusGroupType.DataRegion);
+ for (Pair<Long, TConsensusGroupId> pair : regionSlotsCounter) {
+ counter.put(pair.getRight(), pair.getLeft().intValue());
+ }
+
+ DataPartitionPolicyTable allotTable = dataAllotTableMap.get(database);
+ allotTable.acquireLock();
+ DataPartitionTable dataPartitionTable = new DataPartitionTable();
+
+ try {
+ // Enumerate SeriesPartitionSlot
+ for (Map.Entry<TSeriesPartitionSlot, TTimeSlotList> seriesPartitionEntry :
+ unassignedPartitionSlotsMap.entrySet()) {
+ SeriesPartitionTable seriesPartitionTable = new SeriesPartitionTable();
+
+ // Enumerate TimePartitionSlot in ascending order
+ TSeriesPartitionSlot seriesPartitionSlot = seriesPartitionEntry.getKey();
+ List<TTimePartitionSlot> timePartitionSlots =
+ seriesPartitionEntry.getValue().getTimePartitionSlots();
+ timePartitionSlots.sort(Comparator.comparingLong(TTimePartitionSlot::getStartTime));
+
+ for (TTimePartitionSlot timePartitionSlot : timePartitionSlots) {
+
+ // 1. The historical DataPartition will try to inherit successor DataPartition first
+ TConsensusGroupId successor =
+ getPartitionManager()
+ .getSuccessorDataPartition(database, seriesPartitionSlot, timePartitionSlot);
+ if (successor != null && counter.containsKey(successor)) {
Review Comment:
Yep. Only Running, Available and Discouraged RegionGroups are inside the counter, which is guaranteed by the PartitionMananger.getSortedRegionGroupSlotsCounter()
--
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
[GitHub] [iotdb] CRZbulabula commented on a diff in pull request #10648: [IOTDB-6079] Cluster computing resource balance
Posted by "CRZbulabula (via GitHub)" <gi...@apache.org>.
CRZbulabula commented on code in PR #10648:
URL: https://github.com/apache/iotdb/pull/10648#discussion_r1278897190
##########
integration-test/src/test/java/org/apache/iotdb/session/it/IoTDBSessionSchemaTemplateIT.java:
##########
@@ -471,18 +471,22 @@ public void testHybridAutoExtendSchemaTemplate()
Assert.assertTrue(expectedSeries.isEmpty());
- deviceIds = Arrays.asList("root.db.v4.d1", "root.db.v4.d2");
- timestamps = Arrays.asList(1L, 1L);
measurements = Arrays.asList("a", "b", "c");
- allMeasurements = Arrays.asList(measurements, measurements);
- allTsDataTypes =
- Arrays.asList(
- Arrays.asList(TSDataType.FLOAT, TSDataType.FLOAT, TSDataType.TEXT),
- Arrays.asList(TSDataType.DOUBLE, TSDataType.DOUBLE, TSDataType.INT32));
- allValues = Arrays.asList(Arrays.asList(1f, 2f, "3"), Arrays.asList(1d, 2d, 3));
try {
- session.insertRecords(deviceIds, timestamps, allMeasurements, allTsDataTypes, allValues);
+ session.insertRecord(
Review Comment:
The new partition policy is incompatible with this IT. Updated by @Cpaulyz.
--
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
[GitHub] [iotdb] CRZbulabula commented on a diff in pull request #10648: [IOTDB-6079] Cluster computing resource balance
Posted by "CRZbulabula (via GitHub)" <gi...@apache.org>.
CRZbulabula commented on code in PR #10648:
URL: https://github.com/apache/iotdb/pull/10648#discussion_r1278895048
##########
integration-test/src/test/java/org/apache/iotdb/confignode/it/partition/IoTDBPartitionCreationIT.java:
##########
@@ -419,182 +419,6 @@ public void testPartitionAllocation() throws Exception {
4 + testTimePartitionBatchSize,
testTimePartitionInterval,
dataPartitionTableResp.getDataPartitionTable());
-
Review Comment:
These deleted IT are testing the priority when some RegionGroups aren't in the Running Status, which conflicts with the current partition allocation policy. The other tests in this test class has tested the exception state of Regions.
--
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
[GitHub] [iotdb] CRZbulabula commented on a diff in pull request #10648: [IOTDB-6079] Cluster computing resource balance
Posted by "CRZbulabula (via GitHub)" <gi...@apache.org>.
CRZbulabula commented on code in PR #10648:
URL: https://github.com/apache/iotdb/pull/10648#discussion_r1278947735
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/PartitionBalancer.java:
##########
@@ -16,58 +16,237 @@
* specific language governing permissions and limitations
* under the License.
*/
+
package org.apache.iotdb.confignode.manager.load.balancer;
+import org.apache.iotdb.common.rpc.thrift.TConsensusGroupId;
+import org.apache.iotdb.common.rpc.thrift.TConsensusGroupType;
import org.apache.iotdb.common.rpc.thrift.TSeriesPartitionSlot;
+import org.apache.iotdb.common.rpc.thrift.TTimePartitionSlot;
import org.apache.iotdb.commons.partition.DataPartitionTable;
import org.apache.iotdb.commons.partition.SchemaPartitionTable;
+import org.apache.iotdb.commons.partition.SeriesPartitionTable;
+import org.apache.iotdb.commons.structure.BalanceTreeMap;
+import org.apache.iotdb.confignode.exception.DatabaseNotExistsException;
import org.apache.iotdb.confignode.exception.NoAvailableRegionGroupException;
import org.apache.iotdb.confignode.manager.IManager;
-import org.apache.iotdb.confignode.manager.load.balancer.partition.GreedyPartitionAllocator;
-import org.apache.iotdb.confignode.manager.load.balancer.partition.IPartitionAllocator;
+import org.apache.iotdb.confignode.manager.load.balancer.partition.DataPartitionPolicyTable;
+import org.apache.iotdb.confignode.manager.partition.PartitionManager;
+import org.apache.iotdb.confignode.manager.schema.ClusterSchemaManager;
import org.apache.iotdb.confignode.rpc.thrift.TTimeSlotList;
+import org.apache.iotdb.tsfile.utils.Pair;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import java.util.Comparator;
import java.util.List;
import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
/**
* The SeriesPartitionSlotBalancer provides interfaces to generate optimal Partition allocation and
* migration plans
*/
public class PartitionBalancer {
+ private static final Logger LOGGER = LoggerFactory.getLogger(PartitionBalancer.class);
+
private final IManager configManager;
+ // Map<DatabaseName, DataPartitionPolicyTable>
+ private final Map<String, DataPartitionPolicyTable> dataAllotTableMap;
+
public PartitionBalancer(IManager configManager) {
this.configManager = configManager;
+ this.dataAllotTableMap = new ConcurrentHashMap<>();
}
/**
* Allocate SchemaPartitions
*
* @param unassignedSchemaPartitionSlotsMap SchemaPartitionSlots that should be assigned
- * @return Map<StorageGroupName, SchemaPartitionTable>, the allocating result
+ * @return Map<DatabaseName, SchemaPartitionTable>, the allocating result
*/
public Map<String, SchemaPartitionTable> allocateSchemaPartition(
Map<String, List<TSeriesPartitionSlot>> unassignedSchemaPartitionSlotsMap)
throws NoAvailableRegionGroupException {
- return genPartitionAllocator().allocateSchemaPartition(unassignedSchemaPartitionSlotsMap);
+ Map<String, SchemaPartitionTable> result = new ConcurrentHashMap<>();
+
+ for (Map.Entry<String, List<TSeriesPartitionSlot>> slotsMapEntry :
+ unassignedSchemaPartitionSlotsMap.entrySet()) {
+ final String database = slotsMapEntry.getKey();
+ final List<TSeriesPartitionSlot> unassignedPartitionSlots = slotsMapEntry.getValue();
+
+ // Filter available SchemaRegionGroups and
+ // sort them by the number of allocated SchemaPartitions
+ BalanceTreeMap<TConsensusGroupId, Integer> counter = new BalanceTreeMap<>();
+ List<Pair<Long, TConsensusGroupId>> regionSlotsCounter =
+ getPartitionManager()
+ .getSortedRegionGroupSlotsCounter(database, TConsensusGroupType.SchemaRegion);
+ for (Pair<Long, TConsensusGroupId> pair : regionSlotsCounter) {
+ counter.put(pair.getRight(), pair.getLeft().intValue());
+ }
+
+ // Enumerate SeriesPartitionSlot
+ Map<TSeriesPartitionSlot, TConsensusGroupId> schemaPartitionMap = new ConcurrentHashMap<>();
+ for (TSeriesPartitionSlot seriesPartitionSlot : unassignedPartitionSlots) {
+ // Greedy allocation: allocate the unassigned SchemaPartition to
+ // the RegionGroup whose allocated SchemaPartitions is the least
+ TConsensusGroupId consensusGroupId = counter.getKeyWithMinValue();
+ schemaPartitionMap.put(seriesPartitionSlot, consensusGroupId);
+ counter.put(consensusGroupId, counter.get(consensusGroupId) + 1);
+ }
+ result.put(database, new SchemaPartitionTable(schemaPartitionMap));
+ }
+
+ return result;
}
/**
* Allocate DataPartitions
*
* @param unassignedDataPartitionSlotsMap DataPartitionSlots that should be assigned
- * @return Map<StorageGroupName, DataPartitionTable>, the allocating result
+ * @return Map<DatabaseName, DataPartitionTable>, the allocating result
*/
public Map<String, DataPartitionTable> allocateDataPartition(
Map<String, Map<TSeriesPartitionSlot, TTimeSlotList>> unassignedDataPartitionSlotsMap)
throws NoAvailableRegionGroupException {
- return genPartitionAllocator().allocateDataPartition(unassignedDataPartitionSlotsMap);
+ Map<String, DataPartitionTable> result = new ConcurrentHashMap<>();
+
+ for (Map.Entry<String, Map<TSeriesPartitionSlot, TTimeSlotList>> slotsMapEntry :
+ unassignedDataPartitionSlotsMap.entrySet()) {
+ final String database = slotsMapEntry.getKey();
+ final Map<TSeriesPartitionSlot, TTimeSlotList> unassignedPartitionSlotsMap =
+ slotsMapEntry.getValue();
+
+ // Filter available DataRegionGroups and
+ // sort them by the number of allocated DataPartitions
+ BalanceTreeMap<TConsensusGroupId, Integer> counter = new BalanceTreeMap<>();
+ List<Pair<Long, TConsensusGroupId>> regionSlotsCounter =
+ getPartitionManager()
+ .getSortedRegionGroupSlotsCounter(database, TConsensusGroupType.DataRegion);
+ for (Pair<Long, TConsensusGroupId> pair : regionSlotsCounter) {
+ counter.put(pair.getRight(), pair.getLeft().intValue());
+ }
+
+ DataPartitionPolicyTable allotTable = dataAllotTableMap.get(database);
+ allotTable.acquireLock();
Review Comment:
These lines aren't meet the requirements of try-with-resources
--
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