You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by qi...@apache.org on 2022/06/10 05:41:12 UTC
[iotdb] branch master updated: [IOTDB-3399] Reconstruct PartitionTable (#6199)
This is an automated email from the ASF dual-hosted git repository.
qiaojialin pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new ecf22abdbe [IOTDB-3399] Reconstruct PartitionTable (#6199)
ecf22abdbe is described below
commit ecf22abdbe0d509fe07a5159fdeae66baaa33626
Author: YongzaoDan <33...@users.noreply.github.com>
AuthorDate: Fri Jun 10 13:41:07 2022 +0800
[IOTDB-3399] Reconstruct PartitionTable (#6199)
---
.../request/write/CreateDataPartitionReq.java | 84 +--
.../consensus/request/write/CreateRegionsReq.java | 1 +
.../request/write/CreateSchemaPartitionReq.java | 55 +-
.../consensus/request/write/DeleteRegionsReq.java | 37 +-
.../request/write/DeleteStorageGroupReq.java | 21 +-
.../consensus/response/DataPartitionResp.java | 10 +
.../consensus/response/SchemaPartitionResp.java | 10 +
.../StorageGroupNotExistsException.java} | 9 +-
.../confignode/manager/ClusterSchemaManager.java | 46 +-
.../iotdb/confignode/manager/ConfigManager.java | 12 +-
.../iotdb/confignode/manager/ConsensusManager.java | 2 -
.../iotdb/confignode/manager/PartitionManager.java | 450 +++++++-------
.../iotdb/confignode/manager/load/LoadManager.java | 113 ++--
.../manager/load/balancer/PartitionBalancer.java | 71 +++
.../manager/load/balancer/RegionBalancer.java | 12 +-
.../partition/GreedyPartitionAllocator.java | 124 ++++
.../balancer/partition/IPartitionAllocator.java | 53 ++
.../CopySetRegionAllocator.java | 2 +-
.../{allocator => region}/IRegionAllocator.java | 2 +-
.../confignode/partition/DataPartitionRule.java | 50 --
.../confignode/persistence/ClusterSchemaInfo.java | 229 +-------
.../confignode/persistence/PartitionInfo.java | 645 ---------------------
.../executor/ConfigRequestExecutor.java | 16 +-
.../persistence/partition/PartitionInfo.java | 617 ++++++++++++++++++++
.../persistence/partition/RegionGroup.java | 97 ++++
.../partition/StorageGroupPartitionTable.java | 320 ++++++++++
.../procedure/env/ConfigNodeProcedureEnv.java | 9 +-
.../impl/DeleteStorageGroupProcedure.java | 2 +-
.../thrift/ConfigNodeRPCServiceProcessor.java | 5 -
.../consensus/request/ConfigRequestSerDeTest.java | 93 +--
.../hash/DeviceGroupHashExecutorManualTest.java | 2 +-
.../persistence/ClusterSchemaInfoTest.java | 12 -
.../confignode/persistence/PartitionInfoTest.java | 86 +--
integration/checkstyle.xml | 188 +-----
.../iotdb/commons/partition/DataPartition.java | 199 +------
.../commons/partition/DataPartitionTable.java | 183 ++++++
.../iotdb/commons/partition/SchemaPartition.java | 168 +-----
.../commons/partition/SchemaPartitionTable.java | 177 ++++++
.../commons/partition/SeriesPartitionTable.java | 192 ++++++
.../commons/partition/DataPartitionTableTest.java | 124 ++++
.../iotdb/commons/partition/DataPartitionTest.java | 69 ---
.../partition/SchemaPartitionTableTest.java | 101 ++++
.../commons/partition/SchemaPartitionTest.java | 65 ---
.../partition/SeriesPartitionTableTest.java | 111 ++++
.../utils/ThriftConfigNodeSerDeUtilsTest.java | 14 -
.../java/org/apache/iotdb/rpc/TSStatusCode.java | 10 +-
.../src/main/thrift/confignode.thrift | 2 -
.../java/org/apache/iotdb/tsfile/utils/Pair.java | 8 +
48 files changed, 2625 insertions(+), 2283 deletions(-)
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/CreateDataPartitionReq.java b/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/CreateDataPartitionReq.java
index 914508e792..5ceb01f22b 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/CreateDataPartitionReq.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/CreateDataPartitionReq.java
@@ -18,101 +18,69 @@
*/
package org.apache.iotdb.confignode.consensus.request.write;
-import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet;
-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.utils.BasicStructureSerDeUtil;
-import org.apache.iotdb.commons.utils.ThriftCommonsSerDeUtils;
import org.apache.iotdb.confignode.consensus.request.ConfigRequest;
import org.apache.iotdb.confignode.consensus.request.ConfigRequestType;
+import org.apache.thrift.TException;
+import org.apache.thrift.protocol.TBinaryProtocol;
+import org.apache.thrift.transport.TIOStreamTransport;
+import org.apache.thrift.transport.TTransport;
+
import java.io.DataOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
-import java.util.ArrayList;
import java.util.HashMap;
-import java.util.List;
import java.util.Map;
import java.util.Objects;
/** Create DataPartition by assignedDataPartition */
public class CreateDataPartitionReq extends ConfigRequest {
- private Map<String, Map<TSeriesPartitionSlot, Map<TTimePartitionSlot, List<TRegionReplicaSet>>>>
- assignedDataPartition;
+ private Map<String, DataPartitionTable> assignedDataPartition;
public CreateDataPartitionReq() {
super(ConfigRequestType.CreateDataPartition);
}
- public Map<String, Map<TSeriesPartitionSlot, Map<TTimePartitionSlot, List<TRegionReplicaSet>>>>
- getAssignedDataPartition() {
+ public Map<String, DataPartitionTable> getAssignedDataPartition() {
return assignedDataPartition;
}
- public void setAssignedDataPartition(
- Map<String, Map<TSeriesPartitionSlot, Map<TTimePartitionSlot, List<TRegionReplicaSet>>>>
- assignedDataPartition) {
+ public void setAssignedDataPartition(Map<String, DataPartitionTable> assignedDataPartition) {
this.assignedDataPartition = assignedDataPartition;
}
@Override
protected void serializeImpl(DataOutputStream stream) throws IOException {
- stream.writeInt(ConfigRequestType.CreateDataPartition.ordinal());
+ try {
+ TTransport transport = new TIOStreamTransport(stream);
+ TBinaryProtocol protocol = new TBinaryProtocol(transport);
+
+ stream.writeInt(ConfigRequestType.CreateDataPartition.ordinal());
- stream.writeInt(assignedDataPartition.size());
- for (Map.Entry<
- String, Map<TSeriesPartitionSlot, Map<TTimePartitionSlot, List<TRegionReplicaSet>>>>
- seriesPartitionTimePartitionEntry : assignedDataPartition.entrySet()) {
- BasicStructureSerDeUtil.write(seriesPartitionTimePartitionEntry.getKey(), stream);
- stream.writeInt(seriesPartitionTimePartitionEntry.getValue().size());
- for (Map.Entry<TSeriesPartitionSlot, Map<TTimePartitionSlot, List<TRegionReplicaSet>>>
- timePartitionEntry : seriesPartitionTimePartitionEntry.getValue().entrySet()) {
- ThriftCommonsSerDeUtils.serializeTSeriesPartitionSlot(timePartitionEntry.getKey(), stream);
- stream.writeInt(timePartitionEntry.getValue().size());
- for (Map.Entry<TTimePartitionSlot, List<TRegionReplicaSet>> regionReplicaSetEntry :
- timePartitionEntry.getValue().entrySet()) {
- ThriftCommonsSerDeUtils.serializeTTimePartitionSlot(
- regionReplicaSetEntry.getKey(), stream);
- stream.writeInt(regionReplicaSetEntry.getValue().size());
- for (TRegionReplicaSet regionReplicaSet : regionReplicaSetEntry.getValue()) {
- ThriftCommonsSerDeUtils.serializeTRegionReplicaSet(regionReplicaSet, stream);
- }
- }
+ stream.writeInt(assignedDataPartition.size());
+ for (Map.Entry<String, DataPartitionTable> dataPartitionTableEntry :
+ assignedDataPartition.entrySet()) {
+ BasicStructureSerDeUtil.write(dataPartitionTableEntry.getKey(), stream);
+ dataPartitionTableEntry.getValue().serialize(stream, protocol);
}
+ } catch (TException e) {
+ throw new IOException(e);
}
}
@Override
protected void deserializeImpl(ByteBuffer buffer) throws IOException {
assignedDataPartition = new HashMap<>();
+
int storageGroupNum = buffer.getInt();
for (int i = 0; i < storageGroupNum; i++) {
- String storageGroupName = BasicStructureSerDeUtil.readString(buffer);
- assignedDataPartition.put(storageGroupName, new HashMap<>());
- int seriesPartitionSlotNum = buffer.getInt();
- for (int j = 0; j < seriesPartitionSlotNum; j++) {
- TSeriesPartitionSlot seriesPartitionSlot =
- ThriftCommonsSerDeUtils.deserializeTSeriesPartitionSlot(buffer);
- assignedDataPartition.get(storageGroupName).put(seriesPartitionSlot, new HashMap<>());
- int timePartitionSlotNum = buffer.getInt();
- for (int k = 0; k < timePartitionSlotNum; k++) {
- TTimePartitionSlot timePartitionSlot =
- ThriftCommonsSerDeUtils.deserializeTTimePartitionSlot(buffer);
- assignedDataPartition
- .get(storageGroupName)
- .get(seriesPartitionSlot)
- .put(timePartitionSlot, new ArrayList<>());
- int regionReplicaSetNum = buffer.getInt();
- for (int l = 0; l < regionReplicaSetNum; l++) {
- assignedDataPartition
- .get(storageGroupName)
- .get(seriesPartitionSlot)
- .get(timePartitionSlot)
- .add(ThriftCommonsSerDeUtils.deserializeTRegionReplicaSet(buffer));
- }
- }
- }
+ String storageGroup = BasicStructureSerDeUtil.readString(buffer);
+ DataPartitionTable dataPartitionTable = new DataPartitionTable();
+ dataPartitionTable.deserialize(buffer);
+ assignedDataPartition.put(storageGroup, dataPartitionTable);
}
}
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/CreateRegionsReq.java b/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/CreateRegionsReq.java
index 00ae5fffed..7c9cb3e16a 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/CreateRegionsReq.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/CreateRegionsReq.java
@@ -37,6 +37,7 @@ import java.util.TreeMap;
/** Create regions for specific StorageGroups */
public class CreateRegionsReq extends ConfigRequest {
+ // Map<StorageGroupName, List<TRegionReplicaSet>>
private final Map<String, List<TRegionReplicaSet>> regionMap;
public CreateRegionsReq() {
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/CreateSchemaPartitionReq.java b/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/CreateSchemaPartitionReq.java
index 1fc8949670..f92f0de952 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/CreateSchemaPartitionReq.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/CreateSchemaPartitionReq.java
@@ -18,55 +18,58 @@
*/
package org.apache.iotdb.confignode.consensus.request.write;
-import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet;
-import org.apache.iotdb.common.rpc.thrift.TSeriesPartitionSlot;
+import org.apache.iotdb.commons.partition.SchemaPartitionTable;
import org.apache.iotdb.commons.utils.BasicStructureSerDeUtil;
-import org.apache.iotdb.commons.utils.ThriftCommonsSerDeUtils;
import org.apache.iotdb.confignode.consensus.request.ConfigRequest;
import org.apache.iotdb.confignode.consensus.request.ConfigRequestType;
+import org.apache.thrift.TException;
+import org.apache.thrift.protocol.TBinaryProtocol;
+import org.apache.thrift.transport.TIOStreamTransport;
+import org.apache.thrift.transport.TTransport;
+
import java.io.DataOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.HashMap;
import java.util.Map;
-import java.util.Map.Entry;
import java.util.Objects;
/** Create SchemaPartition by assignedSchemaPartition */
public class CreateSchemaPartitionReq extends ConfigRequest {
- private Map<String, Map<TSeriesPartitionSlot, TRegionReplicaSet>> assignedSchemaPartition;
+ // TODO: Replace this field whit new SchemaPartition
+ private Map<String, SchemaPartitionTable> assignedSchemaPartition;
public CreateSchemaPartitionReq() {
super(ConfigRequestType.CreateSchemaPartition);
}
- public Map<String, Map<TSeriesPartitionSlot, TRegionReplicaSet>> getAssignedSchemaPartition() {
+ public Map<String, SchemaPartitionTable> getAssignedSchemaPartition() {
return assignedSchemaPartition;
}
public void setAssignedSchemaPartition(
- Map<String, Map<TSeriesPartitionSlot, TRegionReplicaSet>> assignedSchemaPartition) {
+ Map<String, SchemaPartitionTable> assignedSchemaPartition) {
this.assignedSchemaPartition = assignedSchemaPartition;
}
@Override
protected void serializeImpl(DataOutputStream stream) throws IOException {
- stream.writeInt(ConfigRequestType.CreateSchemaPartition.ordinal());
+ try {
+ TTransport transport = new TIOStreamTransport(stream);
+ TBinaryProtocol protocol = new TBinaryProtocol(transport);
+
+ stream.writeInt(ConfigRequestType.CreateSchemaPartition.ordinal());
- stream.writeInt(assignedSchemaPartition.size());
- for (Entry<String, Map<TSeriesPartitionSlot, TRegionReplicaSet>> entry :
- assignedSchemaPartition.entrySet()) {
- String storageGroup = entry.getKey();
- Map<TSeriesPartitionSlot, TRegionReplicaSet> partitionSlots = entry.getValue();
- BasicStructureSerDeUtil.write(storageGroup, stream);
- stream.writeInt(partitionSlots.size());
- partitionSlots.forEach(
- (seriesPartitionSlot, regionReplicaSet) -> {
- ThriftCommonsSerDeUtils.serializeTSeriesPartitionSlot(seriesPartitionSlot, stream);
- ThriftCommonsSerDeUtils.serializeTRegionReplicaSet(regionReplicaSet, stream);
- });
+ stream.writeInt(assignedSchemaPartition.size());
+ for (Map.Entry<String, SchemaPartitionTable> schemaPartitionTableEntry :
+ assignedSchemaPartition.entrySet()) {
+ BasicStructureSerDeUtil.write(schemaPartitionTableEntry.getKey(), stream);
+ schemaPartitionTableEntry.getValue().serialize(stream, protocol);
+ }
+ } catch (TException e) {
+ throw new IOException(e);
}
}
@@ -77,15 +80,9 @@ public class CreateSchemaPartitionReq extends ConfigRequest {
int storageGroupNum = buffer.getInt();
for (int i = 0; i < storageGroupNum; i++) {
String storageGroup = BasicStructureSerDeUtil.readString(buffer);
- assignedSchemaPartition.put(storageGroup, new HashMap<>());
- int seriesPartitionSlotNum = buffer.getInt();
- for (int j = 0; j < seriesPartitionSlotNum; j++) {
- TSeriesPartitionSlot seriesPartitionSlot =
- ThriftCommonsSerDeUtils.deserializeTSeriesPartitionSlot(buffer);
- assignedSchemaPartition
- .get(storageGroup)
- .put(seriesPartitionSlot, ThriftCommonsSerDeUtils.deserializeTRegionReplicaSet(buffer));
- }
+ SchemaPartitionTable schemaPartitionTable = new SchemaPartitionTable();
+ schemaPartitionTable.deserialize(buffer);
+ assignedSchemaPartition.put(storageGroup, schemaPartitionTable);
}
}
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/DeleteRegionsReq.java b/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/DeleteRegionsReq.java
index 1517212b41..ffe2e22a70 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/DeleteRegionsReq.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/DeleteRegionsReq.java
@@ -19,6 +19,7 @@
package org.apache.iotdb.confignode.consensus.request.write;
import org.apache.iotdb.common.rpc.thrift.TConsensusGroupId;
+import org.apache.iotdb.commons.utils.BasicStructureSerDeUtil;
import org.apache.iotdb.commons.utils.ThriftCommonsSerDeUtils;
import org.apache.iotdb.confignode.consensus.request.ConfigRequest;
import org.apache.iotdb.confignode.consensus.request.ConfigRequestType;
@@ -27,33 +28,40 @@ import java.io.DataOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
+import java.util.HashMap;
import java.util.List;
+import java.util.Map;
import java.util.Objects;
public class DeleteRegionsReq extends ConfigRequest {
- private final List<TConsensusGroupId> consensusGroupIds;
+ private final Map<String, List<TConsensusGroupId>> deleteRegionMap;
public DeleteRegionsReq() {
super(ConfigRequestType.DeleteRegions);
- this.consensusGroupIds = new ArrayList<>();
+ this.deleteRegionMap = new HashMap<>();
}
- public void addConsensusGroupId(TConsensusGroupId consensusGroupId) {
- consensusGroupIds.add(consensusGroupId);
+ public void addDeleteRegion(String name, TConsensusGroupId consensusGroupId) {
+ deleteRegionMap.computeIfAbsent(name, empty -> new ArrayList<>()).add(consensusGroupId);
}
- public List<TConsensusGroupId> getConsensusGroupIds() {
- return consensusGroupIds;
+ public Map<String, List<TConsensusGroupId>> getDeleteRegionMap() {
+ return deleteRegionMap;
}
@Override
protected void serializeImpl(DataOutputStream stream) throws IOException {
stream.writeInt(ConfigRequestType.DeleteRegions.ordinal());
- stream.writeInt(consensusGroupIds.size());
- for (TConsensusGroupId consensusGroupId : consensusGroupIds) {
- ThriftCommonsSerDeUtils.serializeTConsensusGroupId(consensusGroupId, stream);
+ stream.writeInt(deleteRegionMap.size());
+ for (Map.Entry<String, List<TConsensusGroupId>> consensusGroupIdsEntry :
+ deleteRegionMap.entrySet()) {
+ BasicStructureSerDeUtil.write(consensusGroupIdsEntry.getKey(), stream);
+ stream.writeInt(consensusGroupIdsEntry.getValue().size());
+ for (TConsensusGroupId consensusGroupId : consensusGroupIdsEntry.getValue()) {
+ ThriftCommonsSerDeUtils.serializeTConsensusGroupId(consensusGroupId, stream);
+ }
}
}
@@ -61,7 +69,12 @@ public class DeleteRegionsReq extends ConfigRequest {
protected void deserializeImpl(ByteBuffer buffer) throws IOException {
int length = buffer.getInt();
for (int i = 0; i < length; i++) {
- consensusGroupIds.add(ThriftCommonsSerDeUtils.deserializeTConsensusGroupId(buffer));
+ String name = BasicStructureSerDeUtil.readString(buffer);
+ deleteRegionMap.put(name, new ArrayList<>());
+ int regionNum = buffer.getInt();
+ for (int j = 0; j < regionNum; j++) {
+ deleteRegionMap.get(name).add(ThriftCommonsSerDeUtils.deserializeTConsensusGroupId(buffer));
+ }
}
}
@@ -70,11 +83,11 @@ public class DeleteRegionsReq extends ConfigRequest {
if (this == o) return true;
if (o == null || getClass() != o.getClass()) return false;
DeleteRegionsReq that = (DeleteRegionsReq) o;
- return consensusGroupIds.equals(that.consensusGroupIds);
+ return deleteRegionMap.equals(that.deleteRegionMap);
}
@Override
public int hashCode() {
- return Objects.hash(consensusGroupIds);
+ return Objects.hash(deleteRegionMap);
}
}
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/DeleteStorageGroupReq.java b/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/DeleteStorageGroupReq.java
index e848a844e2..0baddb62bf 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/DeleteStorageGroupReq.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/DeleteStorageGroupReq.java
@@ -18,10 +18,9 @@
*/
package org.apache.iotdb.confignode.consensus.request.write;
-import org.apache.iotdb.commons.utils.ThriftConfigNodeSerDeUtils;
+import org.apache.iotdb.commons.utils.BasicStructureSerDeUtil;
import org.apache.iotdb.confignode.consensus.request.ConfigRequest;
import org.apache.iotdb.confignode.consensus.request.ConfigRequestType;
-import org.apache.iotdb.confignode.rpc.thrift.TStorageGroupSchema;
import java.io.DataOutputStream;
import java.io.IOException;
@@ -30,30 +29,30 @@ import java.util.Objects;
public class DeleteStorageGroupReq extends ConfigRequest {
- private TStorageGroupSchema storageGroup;
+ private String name;
public DeleteStorageGroupReq() {
super(ConfigRequestType.DeleteStorageGroup);
}
- public DeleteStorageGroupReq(TStorageGroupSchema storageGroup) {
+ public DeleteStorageGroupReq(String name) {
this();
- this.storageGroup = storageGroup;
+ this.name = name;
}
- public TStorageGroupSchema getStorageGroup() {
- return storageGroup;
+ public String getName() {
+ return name;
}
@Override
protected void serializeImpl(DataOutputStream stream) throws IOException {
stream.writeInt(ConfigRequestType.DeleteStorageGroup.ordinal());
- ThriftConfigNodeSerDeUtils.serializeTStorageGroupSchema(storageGroup, stream);
+ BasicStructureSerDeUtil.write(name, stream);
}
@Override
protected void deserializeImpl(ByteBuffer buffer) {
- storageGroup = ThriftConfigNodeSerDeUtils.deserializeTStorageGroupSchema(buffer);
+ name = BasicStructureSerDeUtil.readString(buffer);
}
@Override
@@ -61,11 +60,11 @@ public class DeleteStorageGroupReq extends ConfigRequest {
if (this == o) return true;
if (o == null || getClass() != o.getClass()) return false;
DeleteStorageGroupReq that = (DeleteStorageGroupReq) o;
- return storageGroup.equals(that.storageGroup);
+ return name.equals(that.name);
}
@Override
public int hashCode() {
- return Objects.hash(storageGroup);
+ return Objects.hash(name);
}
}
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/consensus/response/DataPartitionResp.java b/confignode/src/main/java/org/apache/iotdb/confignode/consensus/response/DataPartitionResp.java
index 53f3ed0800..249f0db708 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/consensus/response/DataPartitionResp.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/consensus/response/DataPartitionResp.java
@@ -29,6 +29,8 @@ public class DataPartitionResp implements DataSet {
private TSStatus status;
+ private boolean allPartitionsExist;
+
private DataPartition dataPartition;
public DataPartitionResp() {
@@ -43,6 +45,14 @@ public class DataPartitionResp implements DataSet {
this.status = status;
}
+ public boolean isAllPartitionsExist() {
+ return allPartitionsExist;
+ }
+
+ public void setAllPartitionsExist(boolean allPartitionsExist) {
+ this.allPartitionsExist = allPartitionsExist;
+ }
+
public void setDataPartition(DataPartition dataPartition) {
this.dataPartition = dataPartition;
}
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/consensus/response/SchemaPartitionResp.java b/confignode/src/main/java/org/apache/iotdb/confignode/consensus/response/SchemaPartitionResp.java
index e73865e7d5..e74a5ec826 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/consensus/response/SchemaPartitionResp.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/consensus/response/SchemaPartitionResp.java
@@ -29,6 +29,8 @@ public class SchemaPartitionResp implements DataSet {
private TSStatus status;
+ private boolean allPartitionsExist;
+
private SchemaPartition schemaPartition;
public SchemaPartitionResp() {
@@ -43,6 +45,14 @@ public class SchemaPartitionResp implements DataSet {
this.status = status;
}
+ public boolean isAllPartitionsExist() {
+ return allPartitionsExist;
+ }
+
+ public void setAllPartitionsExist(boolean allPartitionsExist) {
+ this.allPartitionsExist = allPartitionsExist;
+ }
+
public void setSchemaPartition(SchemaPartition schemaPartition) {
this.schemaPartition = schemaPartition;
}
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/SeriesPartitionSlotBalancer.java b/confignode/src/main/java/org/apache/iotdb/confignode/exception/StorageGroupNotExistsException.java
similarity index 75%
rename from confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/SeriesPartitionSlotBalancer.java
rename to confignode/src/main/java/org/apache/iotdb/confignode/exception/StorageGroupNotExistsException.java
index 99225af381..aa1a198496 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/SeriesPartitionSlotBalancer.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/exception/StorageGroupNotExistsException.java
@@ -16,6 +16,11 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.iotdb.confignode.manager.load.balancer;
+package org.apache.iotdb.confignode.exception;
-public class SeriesPartitionSlotBalancer {}
+public class StorageGroupNotExistsException extends ConfigNodeException {
+
+ public StorageGroupNotExistsException(String storageGroup) {
+ super(String.format("StorageGroup: %s doesn't exist.", storageGroup));
+ }
+}
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/manager/ClusterSchemaManager.java b/confignode/src/main/java/org/apache/iotdb/confignode/manager/ClusterSchemaManager.java
index f10051c4ea..0ca17a5e86 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/manager/ClusterSchemaManager.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/manager/ClusterSchemaManager.java
@@ -18,8 +18,6 @@
*/
package org.apache.iotdb.confignode.manager;
-import org.apache.iotdb.common.rpc.thrift.TConsensusGroupId;
-import org.apache.iotdb.common.rpc.thrift.TConsensusGroupType;
import org.apache.iotdb.common.rpc.thrift.TSStatus;
import org.apache.iotdb.commons.exception.IllegalPathException;
import org.apache.iotdb.commons.exception.MetadataException;
@@ -33,6 +31,7 @@ import org.apache.iotdb.confignode.consensus.request.write.SetTTLReq;
import org.apache.iotdb.confignode.consensus.request.write.SetTimePartitionIntervalReq;
import org.apache.iotdb.confignode.consensus.response.CountStorageGroupResp;
import org.apache.iotdb.confignode.consensus.response.StorageGroupSchemaResp;
+import org.apache.iotdb.confignode.exception.StorageGroupNotExistsException;
import org.apache.iotdb.confignode.persistence.ClusterSchemaInfo;
import org.apache.iotdb.confignode.rpc.thrift.TStorageGroupSchema;
import org.apache.iotdb.consensus.common.response.ConsensusReadResponse;
@@ -87,40 +86,15 @@ public class ClusterSchemaManager {
return getConsensusManager().write(deleteStorageGroupReq).getStatus();
}
- /**
- * Only leader use this interface. Get the SchemaRegionGroupIds or DataRegionGroupIds from the
- * specific StorageGroup.
- *
- * @param storageGroup StorageGroupName
- * @param type SchemaRegion or DataRegion
- * @return All SchemaRegionGroupIds when type is SchemaRegion, and all DataRegionGroupIds when
- * type is DataRegion
- */
- public List<TConsensusGroupId> getRegionGroupIds(String storageGroup, TConsensusGroupType type) {
- return clusterSchemaInfo.getRegionGroupIds(storageGroup, type);
- }
-
- /**
- * Only leader use this interface.
- *
- * @param storageGroup StorageGroupName
- * @param type SchemaRegion or DataRegion
- * @return Number of Regions currently owned by the specific StorageGroup
- */
- public int getRegionGroupCount(String storageGroup, TConsensusGroupType type)
- throws MetadataException {
- return clusterSchemaInfo.getRegionGroupCount(storageGroup, type);
- }
-
/**
* Only leader use this interface.
*
* @param storageGroup StorageGroupName
- * @return the matched StorageGroupSchema
- * @throws MetadataException when the specific StorageGroup doesn't exist
+ * @return The specific StorageGroupSchema
+ * @throws StorageGroupNotExistsException When the specific StorageGroup doesn't exist
*/
public TStorageGroupSchema getStorageGroupSchemaByName(String storageGroup)
- throws MetadataException {
+ throws StorageGroupNotExistsException {
return clusterSchemaInfo.getMatchedStorageGroupSchemaByName(storageGroup);
}
@@ -135,18 +109,6 @@ public class ClusterSchemaManager {
return clusterSchemaInfo.getMatchedStorageGroupSchemasByName(rawPathList);
}
- /**
- * Only leader use this interface. Contending the Region allocation particle
- *
- * @param storageGroup The specific StorageGroup
- * @param consensusGroupType SchemaRegion or DataRegion
- * @return True if successfully get the Region allocation particle, false otherwise.
- */
- public boolean getRegionAllocationParticle(
- String storageGroup, TConsensusGroupType consensusGroupType) {
- return clusterSchemaInfo.getRegionAllocationParticle(storageGroup, consensusGroupType);
- }
-
public TSStatus setTTL(SetTTLReq setTTLReq) {
// TODO: Inform DataNodes
return getConsensusManager().write(setTTLReq).getStatus();
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java b/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java
index a6d9797b8d..f695373c21 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java
@@ -19,7 +19,6 @@
package org.apache.iotdb.confignode.manager;
-import org.apache.iotdb.common.rpc.thrift.TConsensusGroupType;
import org.apache.iotdb.common.rpc.thrift.TSStatus;
import org.apache.iotdb.common.rpc.thrift.TSeriesPartitionSlot;
import org.apache.iotdb.commons.conf.CommonDescriptor;
@@ -57,10 +56,10 @@ import org.apache.iotdb.confignode.manager.load.LoadManager;
import org.apache.iotdb.confignode.persistence.AuthorInfo;
import org.apache.iotdb.confignode.persistence.ClusterSchemaInfo;
import org.apache.iotdb.confignode.persistence.NodeInfo;
-import org.apache.iotdb.confignode.persistence.PartitionInfo;
import org.apache.iotdb.confignode.persistence.ProcedureInfo;
import org.apache.iotdb.confignode.persistence.UDFInfo;
import org.apache.iotdb.confignode.persistence.executor.ConfigRequestExecutor;
+import org.apache.iotdb.confignode.persistence.partition.PartitionInfo;
import org.apache.iotdb.confignode.rpc.thrift.TConfigNodeRegisterReq;
import org.apache.iotdb.confignode.rpc.thrift.TConfigNodeRegisterResp;
import org.apache.iotdb.confignode.rpc.thrift.TPermissionInfoResp;
@@ -260,15 +259,6 @@ public class ConfigManager implements Manager {
// remove wild
Map<String, TStorageGroupSchema> deleteStorageSchemaMap =
getClusterSchemaManager().getMatchedStorageGroupSchemasByName(deletedPaths);
- for (Map.Entry<String, TStorageGroupSchema> storageGroupSchemaEntry :
- deleteStorageSchemaMap.entrySet()) {
- String sgName = storageGroupSchemaEntry.getKey();
- TStorageGroupSchema deleteStorageSchema = storageGroupSchemaEntry.getValue();
- deleteStorageSchema.setSchemaRegionGroupIds(
- getClusterSchemaManager().getRegionGroupIds(sgName, TConsensusGroupType.SchemaRegion));
- deleteStorageSchema.setDataRegionGroupIds(
- getClusterSchemaManager().getRegionGroupIds(sgName, TConsensusGroupType.DataRegion));
- }
ArrayList<TStorageGroupSchema> parsedDeleteStorageGroups =
new ArrayList<>(deleteStorageSchemaMap.values());
return procedureManager.deleteStorageGroups(parsedDeleteStorageGroups);
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConsensusManager.java b/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConsensusManager.java
index 918c4b693f..d46e6e66b3 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConsensusManager.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConsensusManager.java
@@ -159,6 +159,4 @@ public class ConsensusManager {
public ConsensusGroupId getConsensusGroupId() {
return consensusGroupId;
}
-
- // TODO: Interfaces for LoadBalancer control
}
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/manager/PartitionManager.java b/confignode/src/main/java/org/apache/iotdb/confignode/manager/PartitionManager.java
index 541573b301..767d186693 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/manager/PartitionManager.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/manager/PartitionManager.java
@@ -26,7 +26,8 @@ import org.apache.iotdb.common.rpc.thrift.TSeriesPartitionSlot;
import org.apache.iotdb.common.rpc.thrift.TTimePartitionSlot;
import org.apache.iotdb.commons.concurrent.IoTDBThreadPoolFactory;
import org.apache.iotdb.commons.concurrent.threadpool.ScheduledExecutorUtil;
-import org.apache.iotdb.commons.exception.MetadataException;
+import org.apache.iotdb.commons.partition.DataPartitionTable;
+import org.apache.iotdb.commons.partition.SchemaPartitionTable;
import org.apache.iotdb.commons.partition.executor.SeriesPartitionExecutor;
import org.apache.iotdb.confignode.client.SyncDataNodeClientPool;
import org.apache.iotdb.confignode.conf.ConfigNodeConf;
@@ -43,9 +44,10 @@ import org.apache.iotdb.confignode.consensus.response.DataPartitionResp;
import org.apache.iotdb.confignode.consensus.response.SchemaNodeManagementResp;
import org.apache.iotdb.confignode.consensus.response.SchemaPartitionResp;
import org.apache.iotdb.confignode.exception.NotEnoughDataNodeException;
+import org.apache.iotdb.confignode.exception.StorageGroupNotExistsException;
import org.apache.iotdb.confignode.exception.TimeoutException;
import org.apache.iotdb.confignode.manager.load.LoadManager;
-import org.apache.iotdb.confignode.persistence.PartitionInfo;
+import org.apache.iotdb.confignode.persistence.partition.PartitionInfo;
import org.apache.iotdb.confignode.rpc.thrift.TStorageGroupSchema;
import org.apache.iotdb.consensus.common.DataSet;
import org.apache.iotdb.consensus.common.response.ConsensusReadResponse;
@@ -56,8 +58,6 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.ArrayList;
-import java.util.Collections;
-import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
@@ -92,262 +92,197 @@ public class PartitionManager {
setSeriesPartitionExecutor();
}
- public ScheduledExecutorService getRegionCleaner() {
- return regionCleaner;
+ /** Construct SeriesPartitionExecutor by iotdb-confignode.propertis */
+ private void setSeriesPartitionExecutor() {
+ ConfigNodeConf conf = ConfigNodeDescriptor.getInstance().getConf();
+ this.executor =
+ SeriesPartitionExecutor.getSeriesPartitionExecutor(
+ conf.getSeriesPartitionExecutorClass(), conf.getSeriesPartitionSlotNum());
}
+ // ======================================================
+ // Consensus read/write interfaces
+ // ======================================================
+
/**
- * Get SchemaPartition
+ * Thread-safely get SchemaPartition
*
- * @param physicalPlan SchemaPartitionPlan with partitionSlotsMap
+ * @param req SchemaPartitionPlan with partitionSlotsMap
* @return SchemaPartitionDataSet that contains only existing SchemaPartition
*/
- public DataSet getSchemaPartition(GetSchemaPartitionReq physicalPlan) {
- SchemaPartitionResp schemaPartitionResp;
- ConsensusReadResponse consensusReadResponse = getConsensusManager().read(physicalPlan);
- schemaPartitionResp = (SchemaPartitionResp) consensusReadResponse.getDataset();
- return schemaPartitionResp;
+ public DataSet getSchemaPartition(GetSchemaPartitionReq req) {
+ return getConsensusManager().read(req).getDataset();
}
/**
- * Get SchemaPartition and create a new one if it does not exist
+ * Thread-safely get DataPartition
*
- * @param physicalPlan SchemaPartitionPlan with partitionSlotsMap
- * @return SchemaPartitionResp with DataPartition and TSStatus. SUCCESS_STATUS if all process
- * finish. NOT_ENOUGH_DATA_NODE if the DataNodes is not enough to create new Regions.
+ * @param req DataPartitionPlan with Map<StorageGroupName, Map<SeriesPartitionSlot,
+ * List<TimePartitionSlot>>>
+ * @return DataPartitionDataSet that contains only existing DataPartition
*/
- public DataSet getOrCreateSchemaPartition(GetOrCreateSchemaPartitionReq physicalPlan) {
- Map<String, List<TSeriesPartitionSlot>> noAssignedSchemaPartitionSlots =
- partitionInfo.filterNoAssignedSchemaPartitionSlots(physicalPlan.getPartitionSlotsMap());
-
- if (noAssignedSchemaPartitionSlots.size() > 0) {
-
- // Make sure each StorageGroup has at least one SchemaRegion
- try {
- checkAndAllocateRegionsIfNecessary(
- new ArrayList<>(noAssignedSchemaPartitionSlots.keySet()),
- TConsensusGroupType.SchemaRegion);
- } catch (NotEnoughDataNodeException e) {
- SchemaPartitionResp resp = new SchemaPartitionResp();
- resp.setStatus(new TSStatus(TSStatusCode.NOT_ENOUGH_DATA_NODE.getStatusCode()));
- resp.getStatus()
- .setMessage(
- "ConfigNode failed to allocate DataPartition because there are not enough DataNodes");
- return resp;
- } catch (TimeoutException e) {
- SchemaPartitionResp resp = new SchemaPartitionResp();
- resp.setStatus(new TSStatus(TSStatusCode.TIME_OUT.getStatusCode()));
- resp.getStatus()
- .setMessage(
- "ConfigNode failed to allocate DataPartition because waiting for another thread's Region allocation timeout.");
- return resp;
- } catch (MetadataException e) {
- SchemaPartitionResp resp = new SchemaPartitionResp();
- resp.setStatus(new TSStatus(TSStatusCode.STORAGE_GROUP_NOT_EXIST.getStatusCode()));
- resp.getStatus()
- .setMessage(
- "ConfigNode failed to allocate DataPartition because some StorageGroup doesn't exist.");
- return resp;
- }
-
- // Allocate SchemaPartition
- Map<String, Map<TSeriesPartitionSlot, TRegionReplicaSet>> assignedSchemaPartition =
- allocateSchemaPartition(noAssignedSchemaPartitionSlots);
-
- // Persist SchemaPartition
- CreateSchemaPartitionReq createPlan = new CreateSchemaPartitionReq();
- createPlan.setAssignedSchemaPartition(assignedSchemaPartition);
- getConsensusManager().write(createPlan);
-
- // TODO: Allocate more Regions if necessary
- }
-
- return getSchemaPartition(physicalPlan);
+ public DataSet getDataPartition(GetDataPartitionReq req) {
+ return getConsensusManager().read(req).getDataset();
}
/**
- * TODO: allocate schema partition by LoadManager
+ * Get SchemaPartition and create a new one if it does not exist
*
- * @param noAssignedSchemaPartitionSlotsMap Map<StorageGroupName, List<SeriesPartitionSlot>>
- * @return assign result, Map<StorageGroupName, Map<SeriesPartitionSlot, RegionReplicaSet>>
+ * @param req SchemaPartitionPlan with partitionSlotsMap
+ * @return SchemaPartitionResp with DataPartition and TSStatus. SUCCESS_STATUS if all process
+ * finish. NOT_ENOUGH_DATA_NODE if the DataNodes is not enough to create new Regions. TIME_OUT
+ * if waiting other threads to create Regions for too long. STORAGE_GROUP_NOT_EXIST if some
+ * StorageGroup doesn't exist.
*/
- private Map<String, Map<TSeriesPartitionSlot, TRegionReplicaSet>> allocateSchemaPartition(
- Map<String, List<TSeriesPartitionSlot>> noAssignedSchemaPartitionSlotsMap) {
-
- Map<String, Map<TSeriesPartitionSlot, TRegionReplicaSet>> result = new HashMap<>();
- Map<TConsensusGroupId, TRegionReplicaSet> regionReplicaMap =
- partitionInfo.getRegionReplicaMap();
-
- for (String storageGroup : noAssignedSchemaPartitionSlotsMap.keySet()) {
-
- List<TSeriesPartitionSlot> noAssignedPartitionSlots =
- noAssignedSchemaPartitionSlotsMap.get(storageGroup);
- // List<Pair<allocatedSlotsNum, TConsensusGroupId>>
- List<Pair<Long, TConsensusGroupId>> regionSlotsCounter =
- partitionInfo.getSortedRegionSlotsCounter(
- getClusterSchemaManager()
- .getRegionGroupIds(storageGroup, TConsensusGroupType.SchemaRegion));
-
- Map<TSeriesPartitionSlot, TRegionReplicaSet> allocateResult = new HashMap<>();
- for (TSeriesPartitionSlot seriesPartitionSlot : noAssignedPartitionSlots) {
- // Do greedy allocation
- Pair<Long, TConsensusGroupId> bestRegion = regionSlotsCounter.get(0);
- allocateResult.put(seriesPartitionSlot, regionReplicaMap.get(bestRegion.getRight()));
-
- // Bubble sort
- int index = 0;
- regionSlotsCounter.set(0, new Pair<>(bestRegion.getLeft() + 1, bestRegion.getRight()));
- while (index < regionSlotsCounter.size() - 1
- && regionSlotsCounter.get(index).getLeft()
- > regionSlotsCounter.get(index + 1).getLeft()) {
- Collections.swap(regionSlotsCounter, index, index + 1);
- index += 1;
- }
- }
+ public DataSet getOrCreateSchemaPartition(GetOrCreateSchemaPartitionReq req) {
+ // After all the SchemaPartitions are allocated,
+ // all the read requests about SchemaPartitionTable are parallel.
+ SchemaPartitionResp resp = (SchemaPartitionResp) getSchemaPartition(req);
+ if (resp.isAllPartitionsExist()) {
+ return resp;
+ }
- result.put(storageGroup, allocateResult);
+ // Otherwise, fist ensure that each StorageGroup has at least one SchemaRegion.
+ // This block of code is still parallel and concurrent safe.
+ // Thus, we can prepare the SchemaRegions with maximum efficiency.
+ try {
+ checkAndAllocateRegionsIfNecessary(
+ new ArrayList<>(req.getPartitionSlotsMap().keySet()), TConsensusGroupType.SchemaRegion);
+ } catch (NotEnoughDataNodeException e) {
+ resp.setStatus(
+ new TSStatus(TSStatusCode.NOT_ENOUGH_DATA_NODE.getStatusCode())
+ .setMessage(
+ "ConfigNode failed to allocate DataPartition because there are not enough DataNodes"));
+ return resp;
+ } catch (TimeoutException e) {
+ resp.setStatus(
+ new TSStatus(TSStatusCode.TIME_OUT.getStatusCode())
+ .setMessage(
+ "ConfigNode failed to allocate DataPartition because waiting for another thread's Region allocation timeout."));
+ return resp;
+ } catch (StorageGroupNotExistsException e) {
+ resp.setStatus(
+ new TSStatus(TSStatusCode.STORAGE_GROUP_NOT_EXIST.getStatusCode())
+ .setMessage(
+ "ConfigNode failed to allocate DataPartition because some StorageGroup doesn't exist."));
+ return resp;
}
- return result;
- }
+ // Next, we serialize the creation process of SchemaPartitions to
+ // ensure that each SchemaPartition is created by a unique CreateSchemaPartitionReq.
+ // Because the number of SchemaPartitions per storage group is limited by the number of
+ // SeriesPartitionSlots,
+ // the number of serialized CreateSchemaPartitionReqs is acceptable.
+ synchronized (this) {
+ // Filter unassigned SchemaPartitionSlots
+ Map<String, List<TSeriesPartitionSlot>> unassignedSchemaPartitionSlots =
+ partitionInfo.filterUnassignedSchemaPartitionSlots(req.getPartitionSlotsMap());
+ if (unassignedSchemaPartitionSlots.size() > 0) {
+ // Allocate SchemaPartitions
+ Map<String, SchemaPartitionTable> assignedSchemaPartition =
+ getLoadManager().allocateSchemaPartition(unassignedSchemaPartitionSlots);
+ // Cache allocating result
+ CreateSchemaPartitionReq createPlan = new CreateSchemaPartitionReq();
+ createPlan.setAssignedSchemaPartition(assignedSchemaPartition);
+ getConsensusManager().write(createPlan);
+ }
+ }
- /**
- * Get DataPartition
- *
- * @param physicalPlan DataPartitionPlan with Map<StorageGroupName, Map<SeriesPartitionSlot,
- * List<TimePartitionSlot>>>
- * @return DataPartitionDataSet that contains only existing DataPartition
- */
- public DataSet getDataPartition(GetDataPartitionReq physicalPlan) {
- DataPartitionResp dataPartitionResp;
- ConsensusReadResponse consensusReadResponse = getConsensusManager().read(physicalPlan);
- dataPartitionResp = (DataPartitionResp) consensusReadResponse.getDataset();
- return dataPartitionResp;
+ // TODO: Allocate more SchemaRegions if necessary
+
+ return getSchemaPartition(req);
}
/**
* Get DataPartition and create a new one if it does not exist
*
- * @param physicalPlan DataPartitionPlan with Map<StorageGroupName, Map<SeriesPartitionSlot,
+ * @param req DataPartitionPlan with Map<StorageGroupName, Map<SeriesPartitionSlot,
* List<TimePartitionSlot>>>
* @return DataPartitionResp with DataPartition and TSStatus. SUCCESS_STATUS if all process
- * finish. NOT_ENOUGH_DATA_NODE if the DataNodes is not enough to create new Regions.
+ * finish. NOT_ENOUGH_DATA_NODE if the DataNodes is not enough to create new Regions. TIME_OUT
+ * if waiting other threads to create Regions for too long. STORAGE_GROUP_NOT_EXIST if some
+ * StorageGroup doesn't exist.
*/
- public DataSet getOrCreateDataPartition(GetOrCreateDataPartitionReq physicalPlan) {
- Map<String, Map<TSeriesPartitionSlot, List<TTimePartitionSlot>>> noAssignedDataPartitionSlots =
- partitionInfo.filterNoAssignedDataPartitionSlots(physicalPlan.getPartitionSlotsMap());
+ public DataSet getOrCreateDataPartition(GetOrCreateDataPartitionReq req) {
+ // After all the SchemaPartitions are allocated,
+ // all the read requests about SchemaPartitionTable are parallel.
+ DataPartitionResp resp = (DataPartitionResp) getDataPartition(req);
+ if (resp.isAllPartitionsExist()) {
+ return resp;
+ }
- if (noAssignedDataPartitionSlots.size() > 0) {
+ // Otherwise, fist ensure that each StorageGroup has at least one DataRegion.
+ // This block of code is still parallel and concurrent safe.
+ // Thus, we can prepare the DataRegions with maximum efficiency.
+ try {
+ checkAndAllocateRegionsIfNecessary(
+ new ArrayList<>(req.getPartitionSlotsMap().keySet()), TConsensusGroupType.DataRegion);
+ } catch (NotEnoughDataNodeException e) {
+ resp.setStatus(
+ new TSStatus(TSStatusCode.NOT_ENOUGH_DATA_NODE.getStatusCode())
+ .setMessage(
+ "ConfigNode failed to allocate DataPartition because there are not enough DataNodes"));
+ return resp;
+ } catch (TimeoutException e) {
+ resp.setStatus(
+ new TSStatus(TSStatusCode.TIME_OUT.getStatusCode())
+ .setMessage(
+ "ConfigNode failed to allocate DataPartition because waiting for another thread's Region allocation timeout."));
+ return resp;
+ } catch (StorageGroupNotExistsException e) {
+ resp.setStatus(
+ new TSStatus(TSStatusCode.STORAGE_GROUP_NOT_EXIST.getStatusCode())
+ .setMessage(
+ "ConfigNode failed to allocate DataPartition because some StorageGroup doesn't exist."));
+ return resp;
+ }
- // Make sure each StorageGroup has at least one DataRegion
- try {
- checkAndAllocateRegionsIfNecessary(
- new ArrayList<>(noAssignedDataPartitionSlots.keySet()), TConsensusGroupType.DataRegion);
- } catch (NotEnoughDataNodeException e) {
- DataPartitionResp resp = new DataPartitionResp();
- resp.setStatus(new TSStatus(TSStatusCode.NOT_ENOUGH_DATA_NODE.getStatusCode()));
- resp.getStatus()
- .setMessage(
- "ConfigNode failed to allocate DataPartition because there are not enough DataNodes");
- return resp;
- } catch (TimeoutException e) {
- DataPartitionResp resp = new DataPartitionResp();
- resp.setStatus(new TSStatus(TSStatusCode.TIME_OUT.getStatusCode()));
- resp.getStatus()
- .setMessage(
- "ConfigNode failed to allocate DataPartition because waiting for another thread's Region allocation timeout.");
- return resp;
- } catch (MetadataException e) {
- SchemaPartitionResp resp = new SchemaPartitionResp();
- resp.setStatus(new TSStatus(TSStatusCode.STORAGE_GROUP_NOT_EXIST.getStatusCode()));
- resp.getStatus()
- .setMessage(
- "ConfigNode failed to allocate DataPartition because some StorageGroup doesn't exist.");
- return resp;
+ // Next, we serialize the creation process of DataPartitions to
+ // ensure that each DataPartition is created by a unique CreateDataPartitionReq.
+ // Because the number of DataPartitions per storage group per day is limited by the number of
+ // SeriesPartitionSlots,
+ // the number of serialized CreateDataPartitionReqs is acceptable.
+ synchronized (this) {
+ // Filter unassigned DataPartitionSlots
+ Map<String, Map<TSeriesPartitionSlot, List<TTimePartitionSlot>>>
+ unassignedDataPartitionSlots =
+ partitionInfo.filterUnassignedDataPartitionSlots(req.getPartitionSlotsMap());
+ if (unassignedDataPartitionSlots.size() > 0) {
+ // Allocate DataPartitions
+ Map<String, DataPartitionTable> assignedDataPartition =
+ getLoadManager().allocateDataPartition(unassignedDataPartitionSlots);
+ // Cache allocating result
+ CreateDataPartitionReq createPlan = new CreateDataPartitionReq();
+ createPlan.setAssignedDataPartition(assignedDataPartition);
+ getConsensusManager().write(createPlan);
}
-
- // Allocate DataPartition
- Map<String, Map<TSeriesPartitionSlot, Map<TTimePartitionSlot, List<TRegionReplicaSet>>>>
- assignedDataPartition = allocateDataPartition(noAssignedDataPartitionSlots);
-
- // Persist DataPartition
- CreateDataPartitionReq createPlan = new CreateDataPartitionReq();
- createPlan.setAssignedDataPartition(assignedDataPartition);
- getConsensusManager().write(createPlan);
-
- // TODO: Allocate more Regions if necessary
}
- return getDataPartition(physicalPlan);
+ // TODO: Allocate more Regions if necessary
+
+ return getDataPartition(req);
}
/**
- * TODO: allocate by LoadManager
+ * Check whether each StorageGroup already has a Region of the specified type, and allocate one
+ * Region for each StorageGroup who doesn't have any.
*
- * @param noAssignedDataPartitionSlotsMap Map<StorageGroupName, Map<SeriesPartitionSlot,
- * List<TimePartitionSlot>>>
- * @return assign result, Map<StorageGroupName, Map<SeriesPartitionSlot, Map<TimePartitionSlot,
- * List<RegionReplicaSet>>>>
+ * @param storageGroups List<StorageGroupName>
+ * @param consensusGroupType SchemaRegion or DataRegion
+ * @throws NotEnoughDataNodeException When the number of online DataNodes are too small to
+ * allocate Regions
+ * @throws TimeoutException When waiting other threads to allocate Regions for too long
+ * @throws StorageGroupNotExistsException When some StorageGroups don't exist
*/
- private Map<String, Map<TSeriesPartitionSlot, Map<TTimePartitionSlot, List<TRegionReplicaSet>>>>
- allocateDataPartition(
- Map<String, Map<TSeriesPartitionSlot, List<TTimePartitionSlot>>>
- noAssignedDataPartitionSlotsMap) {
-
- Map<String, Map<TSeriesPartitionSlot, Map<TTimePartitionSlot, List<TRegionReplicaSet>>>>
- result = new HashMap<>();
- Map<TConsensusGroupId, TRegionReplicaSet> regionReplicaMap =
- partitionInfo.getRegionReplicaMap();
-
- for (String storageGroup : noAssignedDataPartitionSlotsMap.keySet()) {
-
- Map<TSeriesPartitionSlot, List<TTimePartitionSlot>> noAssignedPartitionSlotsMap =
- noAssignedDataPartitionSlotsMap.get(storageGroup);
- // List<Pair<allocatedSlotsNum, TConsensusGroupId>>
- List<Pair<Long, TConsensusGroupId>> regionSlotsCounter =
- partitionInfo.getSortedRegionSlotsCounter(
- getClusterSchemaManager()
- .getRegionGroupIds(storageGroup, TConsensusGroupType.DataRegion));
-
- Map<TSeriesPartitionSlot, Map<TTimePartitionSlot, List<TRegionReplicaSet>>> allocateResult =
- new HashMap<>();
- for (Map.Entry<TSeriesPartitionSlot, List<TTimePartitionSlot>> seriesPartitionEntry :
- noAssignedPartitionSlotsMap.entrySet()) {
- allocateResult.put(seriesPartitionEntry.getKey(), new HashMap<>());
- for (TTimePartitionSlot timePartitionSlot : seriesPartitionEntry.getValue()) {
-
- // Do greedy allocation
- Pair<Long, TConsensusGroupId> bestRegion = regionSlotsCounter.get(0);
- allocateResult
- .get(seriesPartitionEntry.getKey())
- .computeIfAbsent(timePartitionSlot, key -> new ArrayList<>())
- .add(regionReplicaMap.get(bestRegion.getRight()));
-
- // Bubble sort
- int index = 0;
- regionSlotsCounter.set(0, new Pair<>(bestRegion.getLeft() + 1, bestRegion.getRight()));
- while (index < regionSlotsCounter.size() - 1
- && regionSlotsCounter.get(index).getLeft()
- > regionSlotsCounter.get(index + 1).getLeft()) {
- Collections.swap(regionSlotsCounter, index, index + 1);
- index += 1;
- }
- }
- }
-
- result.put(storageGroup, allocateResult);
- }
- return result;
- }
-
private void checkAndAllocateRegionsIfNecessary(
List<String> storageGroups, TConsensusGroupType consensusGroupType)
- throws NotEnoughDataNodeException, TimeoutException, MetadataException {
+ throws NotEnoughDataNodeException, TimeoutException, StorageGroupNotExistsException {
int leastDataNode = 0;
List<String> unreadyStorageGroups = new ArrayList<>();
for (String storageGroup : storageGroups) {
- if (getClusterSchemaManager().getRegionGroupCount(storageGroup, consensusGroupType) == 0) {
+ if (getRegionCount(storageGroup, consensusGroupType) == 0) {
+ // Update leastDataNode
TStorageGroupSchema storageGroupSchema =
getClusterSchemaManager().getStorageGroupSchemaByName(storageGroup);
switch (consensusGroupType) {
@@ -358,12 +293,13 @@ public class PartitionManager {
case DataRegion:
leastDataNode = Math.max(leastDataNode, storageGroupSchema.getDataReplicationFactor());
}
+
// Recording StorageGroups without Region
unreadyStorageGroups.add(storageGroup);
}
}
if (getNodeManager().getOnlineDataNodeCount() < leastDataNode) {
- // Make sure DataNode enough
+ // Make sure DataNodes enough
throw new NotEnoughDataNodeException();
}
@@ -371,7 +307,7 @@ public class PartitionManager {
List<String> storageGroupsInAllocation = new ArrayList<>();
for (String storageGroup : unreadyStorageGroups) {
// Try to get the allocation particle
- if (getClusterSchemaManager().getRegionAllocationParticle(storageGroup, consensusGroupType)) {
+ if (partitionInfo.getRegionAllocationParticle(storageGroup, consensusGroupType)) {
storageGroupsNeedAllocation.add(storageGroup);
} else {
storageGroupsInAllocation.add(storageGroup);
@@ -398,15 +334,15 @@ public class PartitionManager {
}
// Do Region allocation for those StorageGroups who get the particle
- // TODO: Put back Region allocation particles when creation process failed. We can use Procedure
- // to protect this
+ // TODO: Use Procedure to put back Region allocation particles when creation process failed.
getLoadManager().initializeRegions(storageGroupsNeedAllocation, consensusGroupType, regionNum);
+ // TODO: Put back particles after creation
// Waiting allocation for those StorageGroups who don't get the particle
- for (int retry = 0; retry < 10; retry++) {
+ for (int retry = 0; retry < 100; retry++) {
boolean allocationFinished = true;
for (String storageGroup : storageGroupsInAllocation) {
- if (getClusterSchemaManager().getRegionGroupCount(storageGroup, consensusGroupType) > 0) {
+ if (getRegionCount(storageGroup, consensusGroupType) == 0) {
allocationFinished = false;
break;
}
@@ -416,8 +352,8 @@ public class PartitionManager {
}
try {
- // Sleep 1s to wait Region allocation
- TimeUnit.SECONDS.sleep(1);
+ // Sleep 200ms to wait Region allocation
+ TimeUnit.MILLISECONDS.sleep(200);
} catch (InterruptedException e) {
LOGGER.warn("The PartitionManager is interrupted.", e);
}
@@ -425,46 +361,52 @@ public class PartitionManager {
throw new TimeoutException("");
}
- /** Get all allocated RegionReplicaSets */
- public List<TRegionReplicaSet> getAllocatedRegions() {
- return partitionInfo.getAllocatedRegions();
- }
+ // ======================================================
+ // Leader scheduling interfaces
+ // ======================================================
- /** Construct SeriesPartitionExecutor by iotdb-confignode.propertis */
- private void setSeriesPartitionExecutor() {
- ConfigNodeConf conf = ConfigNodeDescriptor.getInstance().getConf();
- this.executor =
- SeriesPartitionExecutor.getSeriesPartitionExecutor(
- conf.getSeriesPartitionExecutorClass(), conf.getSeriesPartitionSlotNum());
+ /**
+ * Only leader use this interface
+ *
+ * @return All Regions' RegionReplicaSet
+ */
+ public List<TRegionReplicaSet> getAllReplicaSets() {
+ return partitionInfo.getAllReplicaSets();
}
/**
- * Get TSeriesPartitionSlot
+ * Only leader use this interface. Get the number of Regions currently owned by the specific
+ * StorageGroup
*
- * @param devicePath Full path ending with device name
- * @return SeriesPartitionSlot
+ * @param storageGroup StorageGroupName
+ * @param type SchemaRegion or DataRegion
+ * @return Number of Regions currently owned by the specific StorageGroup
+ * @throws StorageGroupNotExistsException When the specific StorageGroup doesn't exist
*/
- public TSeriesPartitionSlot getSeriesPartitionSlot(String devicePath) {
- return executor.getSeriesPartitionSlot(devicePath);
+ public int getRegionCount(String storageGroup, TConsensusGroupType type)
+ throws StorageGroupNotExistsException {
+ return partitionInfo.getRegionCount(storageGroup, type);
}
/**
* Only leader use this interface.
*
- * @return the next RegionGroupId
+ * @param storageGroup StorageGroupName
+ * @param type SchemaRegion or DataRegion
+ * @return The specific StorageGroup's Regions that sorted by the number of allocated slots
*/
- public int generateNextRegionGroupId() {
- return partitionInfo.generateNextRegionGroupId();
+ public List<Pair<Long, TConsensusGroupId>> getSortedRegionSlotsCounter(
+ String storageGroup, TConsensusGroupType type) {
+ return partitionInfo.getSortedRegionSlotsCounter(storageGroup, type);
}
/**
- * Only leader use this interface.
+ * Only leader use this interface
*
- * @param groupIds List<TConsensusGroupId>
- * @return RegionReplicaSet by the specific TConsensusGroupIds
+ * @return the next RegionGroupId
*/
- public List<TRegionReplicaSet> getRegionReplicaSets(List<TConsensusGroupId> groupIds) {
- return partitionInfo.getRegionReplicaSets(groupIds);
+ public int generateNextRegionGroupId() {
+ return partitionInfo.generateNextRegionGroupId();
}
/**
@@ -506,6 +448,20 @@ public class PartitionManager {
}
}
+ /**
+ * Get TSeriesPartitionSlot
+ *
+ * @param devicePath Full path ending with device name
+ * @return SeriesPartitionSlot
+ */
+ public TSeriesPartitionSlot getSeriesPartitionSlot(String devicePath) {
+ return executor.getSeriesPartitionSlot(devicePath);
+ }
+
+ public ScheduledExecutorService getRegionCleaner() {
+ return regionCleaner;
+ }
+
private ConsensusManager getConsensusManager() {
return configManager.getConsensusManager();
}
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/LoadManager.java b/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/LoadManager.java
index a780b52109..857c188bcc 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/LoadManager.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/LoadManager.java
@@ -23,24 +23,27 @@ import org.apache.iotdb.common.rpc.thrift.TConsensusGroupType;
import org.apache.iotdb.common.rpc.thrift.TDataNodeInfo;
import org.apache.iotdb.common.rpc.thrift.THeartbeatReq;
import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet;
-import org.apache.iotdb.commons.exception.MetadataException;
+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.confignode.client.AsyncDataNodeClientPool;
import org.apache.iotdb.confignode.client.handlers.HeartbeatHandler;
import org.apache.iotdb.confignode.conf.ConfigNodeDescriptor;
import org.apache.iotdb.confignode.consensus.request.write.CreateRegionsReq;
import org.apache.iotdb.confignode.exception.NotEnoughDataNodeException;
+import org.apache.iotdb.confignode.exception.StorageGroupNotExistsException;
import org.apache.iotdb.confignode.manager.ClusterSchemaManager;
import org.apache.iotdb.confignode.manager.ConsensusManager;
import org.apache.iotdb.confignode.manager.Manager;
import org.apache.iotdb.confignode.manager.NodeManager;
+import org.apache.iotdb.confignode.manager.load.balancer.PartitionBalancer;
import org.apache.iotdb.confignode.manager.load.balancer.RegionBalancer;
import org.apache.iotdb.confignode.manager.load.heartbeat.HeartbeatCache;
-import org.apache.iotdb.confignode.rpc.thrift.TStorageGroupSchema;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -61,7 +64,9 @@ public class LoadManager implements Runnable {
ConfigNodeDescriptor.getInstance().getConf().getHeartbeatInterval();
private final HeartbeatCache heartbeatCache;
+ // Balancers
private final RegionBalancer regionBalancer;
+ private final PartitionBalancer partitionBalancer;
private final Map<TConsensusGroupId, TRegionReplicaSet> replicaScoreMap;
@@ -72,6 +77,7 @@ public class LoadManager implements Runnable {
this.heartbeatCache = new HeartbeatCache();
this.regionBalancer = new RegionBalancer(configManager);
+ this.partitionBalancer = new PartitionBalancer(configManager);
this.replicaScoreMap = new TreeMap<>();
}
@@ -86,7 +92,7 @@ public class LoadManager implements Runnable {
*/
public void initializeRegions(
List<String> storageGroups, TConsensusGroupType consensusGroupType, int regionNum)
- throws NotEnoughDataNodeException, MetadataException {
+ throws NotEnoughDataNodeException, StorageGroupNotExistsException {
CreateRegionsReq createRegionsReq =
regionBalancer.genRegionsAllocationPlan(storageGroups, consensusGroupType, regionNum);
createRegionsOnDataNodes(createRegionsReq);
@@ -95,7 +101,7 @@ public class LoadManager implements Runnable {
}
private void createRegionsOnDataNodes(CreateRegionsReq createRegionsReq)
- throws MetadataException {
+ throws StorageGroupNotExistsException {
Map<String, Long> ttlMap = new HashMap<>();
for (String storageGroup : createRegionsReq.getRegionMap().keySet()) {
ttlMap.put(
@@ -105,86 +111,35 @@ public class LoadManager implements Runnable {
AsyncDataNodeClientPool.getInstance().createRegions(createRegionsReq, ttlMap);
}
- private THeartbeatReq genHeartbeatReq() {
- return new THeartbeatReq(System.currentTimeMillis());
+ /**
+ * Allocate SchemaPartitions
+ *
+ * @param unassignedSchemaPartitionSlotsMap SchemaPartitionSlots that should be assigned
+ * @return Map<StorageGroupName, SchemaPartitionTable>, the allocating result
+ */
+ public Map<String, SchemaPartitionTable> allocateSchemaPartition(
+ Map<String, List<TSeriesPartitionSlot>> unassignedSchemaPartitionSlotsMap) {
+ return partitionBalancer.allocateSchemaPartition(unassignedSchemaPartitionSlotsMap);
}
- private void regionExpansion() {
- // Currently, we simply expand the number of regions held by each storage group to
- // 50% of the total CPU cores to facilitate performance testing of multiple regions
-
- int totalCoreNum = 0;
- List<TDataNodeInfo> dataNodeInfos = getNodeManager().getOnlineDataNodes(-1);
- for (TDataNodeInfo dataNodeInfo : dataNodeInfos) {
- totalCoreNum += dataNodeInfo.getCpuCoreNum();
- }
-
- List<String> storageGroups = getClusterSchemaManager().getStorageGroupNames();
- for (String storageGroup : storageGroups) {
- try {
- TStorageGroupSchema storageGroupSchema =
- getClusterSchemaManager().getStorageGroupSchemaByName(storageGroup);
- int totalReplicaNum =
- storageGroupSchema.getSchemaReplicationFactor()
- * storageGroupSchema.getSchemaRegionGroupIdsSize()
- + storageGroupSchema.getDataReplicationFactor()
- * storageGroupSchema.getDataRegionGroupIdsSize();
-
- if (totalReplicaNum < totalCoreNum * 0.5) {
- // Allocate more Regions
- CreateRegionsReq createRegionsReq = null;
-
- // Assume that cluster will get the best efficiency when SchemaRegion:DataRegion is 1:5
- // TODO: Find an optimal SchemaRegion:DataRegion rate.
- if (storageGroupSchema.getSchemaRegionGroupIdsSize() * 5
- > storageGroupSchema.getDataRegionGroupIdsSize()
- && storageGroupSchema.getDataRegionGroupIdsSize()
- < storageGroupSchema.getMaximumDataRegionCount()) {
- // Allocate more DataRegions
-
- // regionNum equals to min(remain cpu core,
- // min(SchemaRegionCnt * 5 - DataRegionCnt, MaxDataRegionCnt - DataRegionCnt))
- int regionNum =
- Math.min(
- ((int) (totalCoreNum * 0.5) - totalReplicaNum)
- / storageGroupSchema.getDataReplicationFactor(),
- Math.min(
- storageGroupSchema.getSchemaRegionGroupIdsSize() * 5
- - storageGroupSchema.getDataRegionGroupIdsSize(),
- storageGroupSchema.getMaximumDataRegionCount()
- - storageGroupSchema.getDataRegionGroupIdsSize()));
-
- createRegionsReq =
- regionBalancer.genRegionsAllocationPlan(
- Collections.singletonList(storageGroup),
- TConsensusGroupType.DataRegion,
- regionNum);
- } else if (storageGroupSchema.getSchemaRegionGroupIdsSize() * 5
- <= storageGroupSchema.getDataRegionGroupIdsSize()
- && storageGroupSchema.getSchemaRegionGroupIdsSize()
- < storageGroupSchema.getMaximumSchemaRegionCount()) {
- // Allocate one more SchemaRegion
- createRegionsReq =
- regionBalancer.genRegionsAllocationPlan(
- Collections.singletonList(storageGroup), TConsensusGroupType.SchemaRegion, 1);
- }
+ /**
+ * Allocate DataPartitions
+ *
+ * @param unassignedDataPartitionSlotsMap DataPartitionSlots that should be assigned
+ * @return Map<StorageGroupName, DataPartitionTable>, the allocating result
+ */
+ public Map<String, DataPartitionTable> allocateDataPartition(
+ Map<String, Map<TSeriesPartitionSlot, List<TTimePartitionSlot>>>
+ unassignedDataPartitionSlotsMap) {
+ return partitionBalancer.allocateDataPartition(unassignedDataPartitionSlotsMap);
+ }
- // TODO: use procedure to protect this
- if (createRegionsReq != null) {
- createRegionsOnDataNodes(createRegionsReq);
- getConsensusManager().write(createRegionsReq);
- }
- }
- } catch (MetadataException e) {
- LOGGER.warn("Meet error when doing regionExpansion", e);
- } catch (NotEnoughDataNodeException ignore) {
- // The LoadManager will expand Regions automatically after there are enough DataNodes.
- }
- }
+ private THeartbeatReq genHeartbeatReq() {
+ return new THeartbeatReq(System.currentTimeMillis());
}
private void doLoadBalancing() {
- regionExpansion();
+ // regionExpansion();
// TODO: update replicaScoreMap
}
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/PartitionBalancer.java b/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/PartitionBalancer.java
new file mode 100644
index 0000000000..88c4d1bcee
--- /dev/null
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/PartitionBalancer.java
@@ -0,0 +1,71 @@
+/*
+ * 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;
+
+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.confignode.manager.Manager;
+import org.apache.iotdb.confignode.manager.load.balancer.partition.GreedyPartitionAllocator;
+import org.apache.iotdb.confignode.manager.load.balancer.partition.IPartitionAllocator;
+
+import java.util.List;
+import java.util.Map;
+
+/**
+ * The SeriesPartitionSlotBalancer provides interfaces to generate optimal Partition allocation and
+ * migration plans
+ */
+public class PartitionBalancer {
+
+ private final Manager configManager;
+
+ public PartitionBalancer(Manager configManager) {
+ this.configManager = configManager;
+ }
+
+ /**
+ * Allocate SchemaPartitions
+ *
+ * @param unassignedSchemaPartitionSlotsMap SchemaPartitionSlots that should be assigned
+ * @return Map<StorageGroupName, SchemaPartitionTable>, the allocating result
+ */
+ public Map<String, SchemaPartitionTable> allocateSchemaPartition(
+ Map<String, List<TSeriesPartitionSlot>> unassignedSchemaPartitionSlotsMap) {
+ return genPartitionAllocator().allocateSchemaPartition(unassignedSchemaPartitionSlotsMap);
+ }
+
+ /**
+ * Allocate DataPartitions
+ *
+ * @param unassignedDataPartitionSlotsMap DataPartitionSlots that should be assigned
+ * @return Map<StorageGroupName, DataPartitionTable>, the allocating result
+ */
+ public Map<String, DataPartitionTable> allocateDataPartition(
+ Map<String, Map<TSeriesPartitionSlot, List<TTimePartitionSlot>>>
+ unassignedDataPartitionSlotsMap) {
+ return genPartitionAllocator().allocateDataPartition(unassignedDataPartitionSlotsMap);
+ }
+
+ private IPartitionAllocator genPartitionAllocator() {
+ // TODO: The type of PartitionAllocator should be configurable
+ return new GreedyPartitionAllocator(configManager);
+ }
+}
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/RegionBalancer.java b/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/RegionBalancer.java
index 40d66f01a1..79e6656940 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/RegionBalancer.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/RegionBalancer.java
@@ -22,15 +22,15 @@ import org.apache.iotdb.common.rpc.thrift.TConsensusGroupId;
import org.apache.iotdb.common.rpc.thrift.TConsensusGroupType;
import org.apache.iotdb.common.rpc.thrift.TDataNodeInfo;
import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet;
-import org.apache.iotdb.commons.exception.MetadataException;
import org.apache.iotdb.confignode.consensus.request.write.CreateRegionsReq;
import org.apache.iotdb.confignode.exception.NotEnoughDataNodeException;
+import org.apache.iotdb.confignode.exception.StorageGroupNotExistsException;
import org.apache.iotdb.confignode.manager.ClusterSchemaManager;
import org.apache.iotdb.confignode.manager.Manager;
import org.apache.iotdb.confignode.manager.NodeManager;
import org.apache.iotdb.confignode.manager.PartitionManager;
-import org.apache.iotdb.confignode.manager.load.balancer.allocator.CopySetRegionAllocator;
-import org.apache.iotdb.confignode.manager.load.balancer.allocator.IRegionAllocator;
+import org.apache.iotdb.confignode.manager.load.balancer.region.CopySetRegionAllocator;
+import org.apache.iotdb.confignode.manager.load.balancer.region.IRegionAllocator;
import org.apache.iotdb.confignode.rpc.thrift.TStorageGroupSchema;
import java.util.List;
@@ -54,16 +54,16 @@ public class RegionBalancer {
* @param regionNum Number of Regions to be allocated per StorageGroup
* @return CreateRegionsReq
* @throws NotEnoughDataNodeException When the number of DataNodes is not enough for allocation
- * @throws MetadataException When some StorageGroups don't exist
+ * @throws StorageGroupNotExistsException When some StorageGroups don't exist
*/
public CreateRegionsReq genRegionsAllocationPlan(
List<String> storageGroups, TConsensusGroupType consensusGroupType, int regionNum)
- throws NotEnoughDataNodeException, MetadataException {
+ throws NotEnoughDataNodeException, StorageGroupNotExistsException {
CreateRegionsReq createRegionsReq = new CreateRegionsReq();
IRegionAllocator regionAllocator = genRegionAllocator();
List<TDataNodeInfo> onlineDataNodes = getNodeManager().getOnlineDataNodes(-1);
- List<TRegionReplicaSet> allocatedRegions = getPartitionManager().getAllocatedRegions();
+ List<TRegionReplicaSet> allocatedRegions = getPartitionManager().getAllReplicaSets();
for (String storageGroup : storageGroups) {
// Get schema
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/partition/GreedyPartitionAllocator.java b/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/partition/GreedyPartitionAllocator.java
new file mode 100644
index 0000000000..9a702a3773
--- /dev/null
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/partition/GreedyPartitionAllocator.java
@@ -0,0 +1,124 @@
+/*
+ * 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.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.confignode.manager.Manager;
+import org.apache.iotdb.confignode.manager.PartitionManager;
+import org.apache.iotdb.tsfile.utils.Pair;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+/** Allocating new Partitions by greedy algorithm */
+public class GreedyPartitionAllocator implements IPartitionAllocator {
+
+ private final Manager configManager;
+
+ public GreedyPartitionAllocator(Manager configManager) {
+ this.configManager = configManager;
+ }
+
+ @Override
+ public Map<String, SchemaPartitionTable> allocateSchemaPartition(
+ Map<String, List<TSeriesPartitionSlot>> unassignedSchemaPartitionSlotsMap) {
+ Map<String, SchemaPartitionTable> result = new ConcurrentHashMap<>();
+
+ unassignedSchemaPartitionSlotsMap.forEach(
+ (storageGroup, unassignedPartitionSlots) -> {
+ // List<Pair<allocatedSlotsNum, TConsensusGroupId>>
+ List<Pair<Long, TConsensusGroupId>> regionSlotsCounter =
+ getPartitionManager()
+ .getSortedRegionSlotsCounter(storageGroup, TConsensusGroupType.SchemaRegion);
+
+ // Enumerate SeriesPartitionSlot
+ Map<TSeriesPartitionSlot, TConsensusGroupId> schemaPartitionMap =
+ new ConcurrentHashMap<>();
+ for (TSeriesPartitionSlot seriesPartitionSlot : unassignedPartitionSlots) {
+ // Greedy allocation
+ schemaPartitionMap.put(seriesPartitionSlot, regionSlotsCounter.get(0).getRight());
+ // Bubble sort
+ bubbleSort(regionSlotsCounter);
+ }
+ result.put(storageGroup, new SchemaPartitionTable(schemaPartitionMap));
+ });
+
+ return result;
+ }
+
+ @Override
+ public Map<String, DataPartitionTable> allocateDataPartition(
+ Map<String, Map<TSeriesPartitionSlot, List<TTimePartitionSlot>>>
+ unassignedDataPartitionSlotsMap) {
+ Map<String, DataPartitionTable> result = new ConcurrentHashMap<>();
+
+ unassignedDataPartitionSlotsMap.forEach(
+ (storageGroup, unassignedPartitionSlotsMap) -> {
+ // List<Pair<allocatedSlotsNum, TConsensusGroupId>>
+ List<Pair<Long, TConsensusGroupId>> regionSlotsCounter =
+ getPartitionManager()
+ .getSortedRegionSlotsCounter(storageGroup, TConsensusGroupType.DataRegion);
+
+ // Enumerate SeriesPartitionSlot
+ Map<TSeriesPartitionSlot, SeriesPartitionTable> dataPartitionMap =
+ new ConcurrentHashMap<>();
+ for (Map.Entry<TSeriesPartitionSlot, List<TTimePartitionSlot>> seriesPartitionEntry :
+ unassignedPartitionSlotsMap.entrySet()) {
+ // Enumerate TimePartitionSlot
+ Map<TTimePartitionSlot, List<TConsensusGroupId>> seriesPartitionMap =
+ new ConcurrentHashMap<>();
+ for (TTimePartitionSlot timePartitionSlot : seriesPartitionEntry.getValue()) {
+ // Greedy allocation
+ seriesPartitionMap.put(
+ timePartitionSlot,
+ Collections.singletonList(regionSlotsCounter.get(0).getRight()));
+ // Bubble sort
+ bubbleSort(regionSlotsCounter);
+ }
+ dataPartitionMap.put(
+ seriesPartitionEntry.getKey(), new SeriesPartitionTable(seriesPartitionMap));
+ }
+ result.put(storageGroup, new DataPartitionTable(dataPartitionMap));
+ });
+
+ return result;
+ }
+
+ private void bubbleSort(List<Pair<Long, TConsensusGroupId>> regionSlotsCounter) {
+ int index = 0;
+ regionSlotsCounter.get(0).setLeft(regionSlotsCounter.get(0).getLeft() + 1);
+ while (index < regionSlotsCounter.size() - 1
+ && regionSlotsCounter.get(index).getLeft() > regionSlotsCounter.get(index + 1).getLeft()) {
+ Collections.swap(regionSlotsCounter, index, index + 1);
+ index += 1;
+ }
+ }
+
+ private PartitionManager getPartitionManager() {
+ return configManager.getPartitionManager();
+ }
+}
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/partition/IPartitionAllocator.java b/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/partition/IPartitionAllocator.java
new file mode 100644
index 0000000000..93d0a40c74
--- /dev/null
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/partition/IPartitionAllocator.java
@@ -0,0 +1,53 @@
+/*
+ * 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.TSeriesPartitionSlot;
+import org.apache.iotdb.common.rpc.thrift.TTimePartitionSlot;
+import org.apache.iotdb.commons.partition.DataPartitionTable;
+import org.apache.iotdb.commons.partition.SchemaPartitionTable;
+
+import java.util.List;
+import java.util.Map;
+
+/**
+ * The IPartitionAllocator is a functional interface, which means a new functional class who
+ * implements the IPartitionAllocator must be created for each Partition allocation.
+ */
+public interface IPartitionAllocator {
+
+ /**
+ * Allocate SchemaPartitions
+ *
+ * @param unassignedSchemaPartitionSlotsMap SchemaPartitionSlots that should be assigned
+ * @return Map<StorageGroupName, SchemaPartitionTable>, the allocating result
+ */
+ Map<String, SchemaPartitionTable> allocateSchemaPartition(
+ Map<String, List<TSeriesPartitionSlot>> unassignedSchemaPartitionSlotsMap);
+
+ /**
+ * Allocate DataPartitions
+ *
+ * @param unassignedDataPartitionSlotsMap DataPartitionSlots that should be assigned
+ * @return Map<StorageGroupName, DataPartitionTable>, the allocating result
+ */
+ Map<String, DataPartitionTable> allocateDataPartition(
+ Map<String, Map<TSeriesPartitionSlot, List<TTimePartitionSlot>>>
+ unassignedDataPartitionSlotsMap);
+}
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/allocator/CopySetRegionAllocator.java b/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/region/CopySetRegionAllocator.java
similarity index 98%
rename from confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/allocator/CopySetRegionAllocator.java
rename to confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/region/CopySetRegionAllocator.java
index b3e5cb30fc..384c919e83 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/allocator/CopySetRegionAllocator.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/region/CopySetRegionAllocator.java
@@ -16,7 +16,7 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.iotdb.confignode.manager.load.balancer.allocator;
+package org.apache.iotdb.confignode.manager.load.balancer.region;
import org.apache.iotdb.common.rpc.thrift.TConsensusGroupId;
import org.apache.iotdb.common.rpc.thrift.TDataNodeInfo;
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/allocator/IRegionAllocator.java b/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/region/IRegionAllocator.java
similarity index 96%
rename from confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/allocator/IRegionAllocator.java
rename to confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/region/IRegionAllocator.java
index 605de85cb4..ead09395cb 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/allocator/IRegionAllocator.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/region/IRegionAllocator.java
@@ -16,7 +16,7 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.iotdb.confignode.manager.load.balancer.allocator;
+package org.apache.iotdb.confignode.manager.load.balancer.region;
import org.apache.iotdb.common.rpc.thrift.TConsensusGroupId;
import org.apache.iotdb.common.rpc.thrift.TDataNodeInfo;
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/partition/DataPartitionRule.java b/confignode/src/main/java/org/apache/iotdb/confignode/partition/DataPartitionRule.java
deleted file mode 100644
index 1593c6e919..0000000000
--- a/confignode/src/main/java/org/apache/iotdb/confignode/partition/DataPartitionRule.java
+++ /dev/null
@@ -1,50 +0,0 @@
-/*
- * 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.partition;
-
-import org.apache.iotdb.tsfile.utils.Pair;
-
-import java.util.ArrayList;
-import java.util.List;
-
-/**
- * DataPartitionRule is used to hold real-time write-load allocation rules i.e. rules = [(0, 0.3),
- * (1, 0.2), (2. 0.5)] means allocate 30% of the write-load to DataRegion-0 and 20% to DataRegion-1
- * and 50% to DataRegion-2
- */
-public class DataPartitionRule {
- // List<Pair<DataRegionID, Write allocation ratio>>
- private final List<Pair<Integer, Double>> rules;
-
- public DataPartitionRule() {
- this.rules = new ArrayList<>();
- }
-
- public DataPartitionRule(List<Pair<Integer, Double>> rules) {
- this.rules = rules;
- }
-
- public void addDataPartitionRule(int dataRegionID, double ratio) {
- this.rules.add(new Pair<>(dataRegionID, ratio));
- }
-
- public List<Pair<Integer, Double>> getDataPartitionRule() {
- return this.rules;
- }
-}
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/persistence/ClusterSchemaInfo.java b/confignode/src/main/java/org/apache/iotdb/confignode/persistence/ClusterSchemaInfo.java
index 0236d6e240..e3bb52823b 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/persistence/ClusterSchemaInfo.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/persistence/ClusterSchemaInfo.java
@@ -18,9 +18,6 @@
*/
package org.apache.iotdb.confignode.persistence;
-import org.apache.iotdb.common.rpc.thrift.TConsensusGroupId;
-import org.apache.iotdb.common.rpc.thrift.TConsensusGroupType;
-import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet;
import org.apache.iotdb.common.rpc.thrift.TSStatus;
import org.apache.iotdb.commons.exception.MetadataException;
import org.apache.iotdb.commons.path.PartialPath;
@@ -28,7 +25,6 @@ import org.apache.iotdb.commons.snapshot.SnapshotProcessor;
import org.apache.iotdb.commons.utils.TestOnly;
import org.apache.iotdb.confignode.consensus.request.read.CountStorageGroupReq;
import org.apache.iotdb.confignode.consensus.request.read.GetStorageGroupReq;
-import org.apache.iotdb.confignode.consensus.request.write.CreateRegionsReq;
import org.apache.iotdb.confignode.consensus.request.write.DeleteStorageGroupReq;
import org.apache.iotdb.confignode.consensus.request.write.SetDataReplicationFactorReq;
import org.apache.iotdb.confignode.consensus.request.write.SetSchemaReplicationFactorReq;
@@ -37,12 +33,11 @@ import org.apache.iotdb.confignode.consensus.request.write.SetTTLReq;
import org.apache.iotdb.confignode.consensus.request.write.SetTimePartitionIntervalReq;
import org.apache.iotdb.confignode.consensus.response.CountStorageGroupResp;
import org.apache.iotdb.confignode.consensus.response.StorageGroupSchemaResp;
+import org.apache.iotdb.confignode.exception.StorageGroupNotExistsException;
import org.apache.iotdb.confignode.rpc.thrift.TStorageGroupSchema;
-import org.apache.iotdb.db.metadata.mnode.StorageGroupMNode;
import org.apache.iotdb.db.metadata.mtree.MTreeAboveSG;
import org.apache.iotdb.rpc.TSStatusCode;
import org.apache.iotdb.tsfile.utils.Pair;
-import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -54,7 +49,6 @@ import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.util.ArrayList;
-import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
@@ -75,16 +69,10 @@ public class ClusterSchemaInfo implements SnapshotProcessor {
private final ReentrantReadWriteLock storageGroupReadWriteLock;
private final MTreeAboveSG mTree;
- // Particle for allocating the first Region
- private final Set<String> schemaRegionParticles;
- private final Set<String> dataRegionParticles;
-
private final String snapshotFileName = "cluster_schema.bin";
public ClusterSchemaInfo() throws IOException {
storageGroupReadWriteLock = new ReentrantReadWriteLock();
- schemaRegionParticles = Collections.synchronizedSet(new HashSet<>());
- dataRegionParticles = Collections.synchronizedSet(new HashSet<>());
try {
mTree = new MTreeAboveSG();
@@ -95,11 +83,11 @@ public class ClusterSchemaInfo implements SnapshotProcessor {
}
/**
- * Persistence new StorageGroupSchema
+ * Cache StorageGroupSchema
*
* @param req SetStorageGroupReq
- * @return SUCCESS_STATUS if the StorageGroup is set successfully. PERSISTENCE_FAILURE if fail to
- * set StorageGroup in MTreeAboveSG.
+ * @return SUCCESS_STATUS if the StorageGroup is set successfully. CACHE_FAILURE if fail to set
+ * StorageGroup in MTreeAboveSG.
*/
public TSStatus setStorageGroup(SetStorageGroupReq req) {
TSStatus result = new TSStatus();
@@ -115,17 +103,11 @@ public class ClusterSchemaInfo implements SnapshotProcessor {
.getStorageGroupNodeByStorageGroupPath(partialPathName)
.setStorageGroupSchema(storageGroupSchema);
- // Mark the new StorageGroup as without Region
- schemaRegionParticles.add(storageGroupSchema.getName());
- dataRegionParticles.add(storageGroupSchema.getName());
-
result.setCode(TSStatusCode.SUCCESS_STATUS.getStatusCode());
-
- LOGGER.info("Successfully set StorageGroup: {}", storageGroupSchema);
} catch (MetadataException e) {
LOGGER.error("Error StorageGroup name", e);
result
- .setCode(TSStatusCode.PERSISTENCE_FAILURE.getStatusCode())
+ .setCode(TSStatusCode.SET_STORAGE_GROUP_FAILED.getStatusCode())
.setMessage("Error StorageGroup name");
} finally {
storageGroupReadWriteLock.writeLock().unlock();
@@ -144,14 +126,10 @@ public class ClusterSchemaInfo implements SnapshotProcessor {
storageGroupReadWriteLock.writeLock().lock();
try {
// Delete StorageGroup
- TStorageGroupSchema storageGroupSchema = req.getStorageGroup();
- PartialPath partialPathName = new PartialPath(storageGroupSchema.getName());
+ String storageGroup = req.getName();
+ PartialPath partialPathName = new PartialPath(storageGroup);
mTree.deleteStorageGroup(partialPathName);
- // Clean the unready set
- schemaRegionParticles.remove(storageGroupSchema.getName());
- dataRegionParticles.remove(storageGroupSchema.getName());
-
result.setCode(TSStatusCode.SUCCESS_STATUS.getStatusCode());
} catch (MetadataException e) {
LOGGER.warn("Storage group not exist", e);
@@ -164,51 +142,6 @@ public class ClusterSchemaInfo implements SnapshotProcessor {
return result;
}
- /**
- * Persistence new RegionGroupIds on specific StorageGroupSchema
- *
- * @param req CreateRegionsReq
- * @return SUCCESS_STATUS if the new RegionGroupIds is persistence successfully.
- * PERSISTENCE_FAILURE if fail to find StorageGroup in MTreeAboveSG.
- */
- public TSStatus createRegions(CreateRegionsReq req) {
- TSStatus result = new TSStatus();
- storageGroupReadWriteLock.writeLock().lock();
-
- try {
- for (Map.Entry<String, List<TRegionReplicaSet>> reqEntry : req.getRegionMap().entrySet()) {
- PartialPath partialPathName = new PartialPath(reqEntry.getKey());
- TStorageGroupSchema storageGroupSchema =
- mTree.getStorageGroupNodeByStorageGroupPath(partialPathName).getStorageGroupSchema();
- reqEntry
- .getValue()
- .forEach(
- regionReplicaSet -> {
- switch (regionReplicaSet.getRegionId().getType()) {
- case SchemaRegion:
- storageGroupSchema
- .getSchemaRegionGroupIds()
- .add(regionReplicaSet.getRegionId());
- break;
- case DataRegion:
- storageGroupSchema
- .getDataRegionGroupIds()
- .add(regionReplicaSet.getRegionId());
- break;
- }
- });
- }
- result.setCode(TSStatusCode.SUCCESS_STATUS.getStatusCode());
- } catch (MetadataException e) {
- LOGGER.error("Error StorageGroup name", e);
- result.setCode(TSStatusCode.PERSISTENCE_FAILURE.getStatusCode());
- } finally {
- storageGroupReadWriteLock.writeLock().unlock();
- }
-
- return result;
- }
-
public TSStatus setTTL(SetTTLReq req) {
TSStatus result = new TSStatus();
storageGroupReadWriteLock.writeLock().lock();
@@ -384,15 +317,17 @@ public class ClusterSchemaInfo implements SnapshotProcessor {
*
* @param storageGroup StorageGroupName
* @return The specific StorageGroupSchema
- * @throws MetadataException from MTree
+ * @throws StorageGroupNotExistsException When the specific StorageGroup doesn't exist
*/
public TStorageGroupSchema getMatchedStorageGroupSchemaByName(String storageGroup)
- throws MetadataException {
+ throws StorageGroupNotExistsException {
storageGroupReadWriteLock.readLock().lock();
try {
return mTree
.getStorageGroupNodeByStorageGroupPath(new PartialPath(storageGroup))
.getStorageGroupSchema();
+ } catch (MetadataException e) {
+ throw new StorageGroupNotExistsException(storageGroup);
} finally {
storageGroupReadWriteLock.readLock().unlock();
}
@@ -420,100 +355,6 @@ public class ClusterSchemaInfo implements SnapshotProcessor {
return schemaMap;
}
- /**
- * Get the SchemaRegionGroupIds or DataRegionGroupIds from the specific StorageGroup.
- *
- * @param storageGroup StorageGroupName
- * @param type SchemaRegion or DataRegion
- * @return All SchemaRegionGroupIds when type is SchemaRegion, and all DataRegionGroupIds when
- * type is DataRegion
- */
- public List<TConsensusGroupId> getRegionGroupIds(String storageGroup, TConsensusGroupType type) {
- List<TConsensusGroupId> result;
- storageGroupReadWriteLock.readLock().lock();
- try {
- StorageGroupMNode mNode =
- (StorageGroupMNode)
- mTree.getStorageGroupNodeByStorageGroupPath(new PartialPath(storageGroup));
- switch (type) {
- case SchemaRegion:
- result = mNode.getStorageGroupSchema().getSchemaRegionGroupIds();
- break;
- case DataRegion:
- result = mNode.getStorageGroupSchema().getDataRegionGroupIds();
- break;
- default:
- result = new ArrayList<>();
- }
- } catch (MetadataException e) {
- LOGGER.error("Error StorageGroup name", e);
- return new ArrayList<>();
- } finally {
- storageGroupReadWriteLock.readLock().unlock();
- }
- return result;
- }
-
- /**
- * Return the number of Regions currently owned by the specific StorageGroup
- *
- * @param storageGroup StorageGroupName
- * @param type SchemaRegion or DataRegion
- * @return Number of Regions currently owned by the specific StorageGroup
- */
- public int getRegionGroupCount(String storageGroup, TConsensusGroupType type)
- throws MetadataException {
- storageGroupReadWriteLock.readLock().lock();
- try {
- StorageGroupMNode mNode =
- (StorageGroupMNode)
- mTree.getStorageGroupNodeByStorageGroupPath(new PartialPath(storageGroup));
- switch (type) {
- case SchemaRegion:
- return mNode.getStorageGroupSchema().getSchemaRegionGroupIdsSize();
- case DataRegion:
- return mNode.getStorageGroupSchema().getDataRegionGroupIdsSize();
- default:
- return 0;
- }
- } finally {
- storageGroupReadWriteLock.readLock().unlock();
- }
- }
-
- /**
- * Contending the Region allocation particle
- *
- * @param storageGroup The specific StorageGroup
- * @param consensusGroupType SchemaRegion or DataRegion
- * @return True if successfully get the Region allocation particle, false otherwise.
- */
- public boolean getRegionAllocationParticle(
- String storageGroup, TConsensusGroupType consensusGroupType) {
- switch (consensusGroupType) {
- case SchemaRegion:
- synchronized (schemaRegionParticles) {
- if (schemaRegionParticles.contains(storageGroup)) {
- schemaRegionParticles.remove(storageGroup);
- return true;
- } else {
- return false;
- }
- }
- case DataRegion:
- synchronized (dataRegionParticles) {
- if (dataRegionParticles.contains(storageGroup)) {
- dataRegionParticles.remove(storageGroup);
- return true;
- } else {
- return false;
- }
- }
- default:
- return false;
- }
- }
-
@Override
public boolean processTakeSnapshot(File snapshotDir) throws IOException {
@@ -531,22 +372,6 @@ public class ClusterSchemaInfo implements SnapshotProcessor {
try {
try (FileOutputStream fileOutputStream = new FileOutputStream(tmpFile);
BufferedOutputStream outputStream = new BufferedOutputStream(fileOutputStream)) {
-
- // Take snapshot for particles
- synchronized (schemaRegionParticles) {
- synchronized (dataRegionParticles) {
- ReadWriteIOUtils.write(schemaRegionParticles.size(), outputStream);
- for (String schemaRegionParticle : schemaRegionParticles) {
- ReadWriteIOUtils.write(schemaRegionParticle, outputStream);
- }
-
- ReadWriteIOUtils.write(dataRegionParticles.size(), outputStream);
- for (String dataRegionParticle : dataRegionParticles) {
- ReadWriteIOUtils.write(dataRegionParticle, outputStream);
- }
- }
- }
-
// Take snapshot for MTree
mTree.serialize(outputStream);
outputStream.flush();
@@ -579,24 +404,6 @@ public class ClusterSchemaInfo implements SnapshotProcessor {
storageGroupReadWriteLock.writeLock().lock();
try (FileInputStream fileInputStream = new FileInputStream(snapshotFile);
BufferedInputStream bufferedInputStream = new BufferedInputStream(fileInputStream)) {
-
- // Load snapshot of particles
- synchronized (schemaRegionParticles) {
- synchronized (dataRegionParticles) {
- int particleNum = ReadWriteIOUtils.readInt(bufferedInputStream);
- schemaRegionParticles.clear();
- for (int i = 0; i < particleNum; i++) {
- schemaRegionParticles.add(ReadWriteIOUtils.readString(bufferedInputStream));
- }
-
- particleNum = ReadWriteIOUtils.readInt(bufferedInputStream);
- dataRegionParticles.clear();
- for (int i = 0; i < particleNum; i++) {
- dataRegionParticles.add(ReadWriteIOUtils.readString(bufferedInputStream));
- }
- }
- }
-
// Load snapshot of MTree
mTree.clear();
mTree.deserialize(bufferedInputStream);
@@ -622,7 +429,7 @@ public class ClusterSchemaInfo implements SnapshotProcessor {
public Pair<Set<String>, Set<PartialPath>> getChildNodePathInNextLevel(PartialPath partialPath) {
Pair<Set<String>, Set<PartialPath>> matchedPathsInNextLevel =
- new Pair(new HashSet<>(), new HashSet<>());
+ new Pair<>(new HashSet<>(), new HashSet<>());
storageGroupReadWriteLock.readLock().lock();
try {
matchedPathsInNextLevel = mTree.getChildNodePathInNextLevel(partialPath);
@@ -636,7 +443,7 @@ public class ClusterSchemaInfo implements SnapshotProcessor {
public Pair<Set<String>, Set<PartialPath>> getChildNodeNameInNextLevel(PartialPath partialPath) {
Pair<Set<String>, Set<PartialPath>> matchedNamesInNextLevel =
- new Pair(new HashSet<>(), new HashSet<>());
+ new Pair<>(new HashSet<>(), new HashSet<>());
storageGroupReadWriteLock.readLock().lock();
try {
matchedNamesInNextLevel = mTree.getChildNodeNameInNextLevel(partialPath);
@@ -648,16 +455,6 @@ public class ClusterSchemaInfo implements SnapshotProcessor {
return matchedNamesInNextLevel;
}
- @TestOnly
- public Set<String> getSchemaRegionParticles() {
- return schemaRegionParticles;
- }
-
- @TestOnly
- public Set<String> getDataRegionParticles() {
- return dataRegionParticles;
- }
-
@TestOnly
public void clear() {
mTree.clear();
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/persistence/PartitionInfo.java b/confignode/src/main/java/org/apache/iotdb/confignode/persistence/PartitionInfo.java
deleted file mode 100644
index 4339f8025d..0000000000
--- a/confignode/src/main/java/org/apache/iotdb/confignode/persistence/PartitionInfo.java
+++ /dev/null
@@ -1,645 +0,0 @@
-/*
- * 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.persistence;
-
-import org.apache.iotdb.common.rpc.thrift.TConsensusGroupId;
-import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet;
-import org.apache.iotdb.common.rpc.thrift.TSStatus;
-import org.apache.iotdb.common.rpc.thrift.TSeriesPartitionSlot;
-import org.apache.iotdb.common.rpc.thrift.TTimePartitionSlot;
-import org.apache.iotdb.commons.partition.DataPartition;
-import org.apache.iotdb.commons.partition.SchemaPartition;
-import org.apache.iotdb.commons.snapshot.SnapshotProcessor;
-import org.apache.iotdb.commons.utils.TestOnly;
-import org.apache.iotdb.confignode.conf.ConfigNodeDescriptor;
-import org.apache.iotdb.confignode.consensus.request.read.GetDataPartitionReq;
-import org.apache.iotdb.confignode.consensus.request.read.GetSchemaPartitionReq;
-import org.apache.iotdb.confignode.consensus.request.write.CreateDataPartitionReq;
-import org.apache.iotdb.confignode.consensus.request.write.CreateRegionsReq;
-import org.apache.iotdb.confignode.consensus.request.write.CreateSchemaPartitionReq;
-import org.apache.iotdb.confignode.consensus.request.write.DeleteRegionsReq;
-import org.apache.iotdb.confignode.consensus.request.write.DeleteStorageGroupReq;
-import org.apache.iotdb.confignode.consensus.request.write.PreDeleteStorageGroupReq;
-import org.apache.iotdb.confignode.consensus.response.DataPartitionResp;
-import org.apache.iotdb.confignode.consensus.response.SchemaNodeManagementResp;
-import org.apache.iotdb.confignode.consensus.response.SchemaPartitionResp;
-import org.apache.iotdb.confignode.rpc.thrift.TStorageGroupSchema;
-import org.apache.iotdb.consensus.common.DataSet;
-import org.apache.iotdb.rpc.TSStatusCode;
-import org.apache.iotdb.tsfile.utils.Pair;
-import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
-
-import org.apache.thrift.TException;
-import org.apache.thrift.protocol.TBinaryProtocol;
-import org.apache.thrift.protocol.TProtocol;
-import org.apache.thrift.transport.TIOStreamTransport;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.File;
-import java.io.FileInputStream;
-import java.io.FileOutputStream;
-import java.io.IOException;
-import java.io.InputStream;
-import java.io.OutputStream;
-import java.util.ArrayList;
-import java.util.Comparator;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-import java.util.UUID;
-import java.util.concurrent.CopyOnWriteArraySet;
-import java.util.concurrent.atomic.AtomicInteger;
-import java.util.concurrent.locks.ReentrantReadWriteLock;
-
-/**
- * The PartitionInfo stores cluster PartitionTable. The PartitionTable including: 1. regionMap:
- * location of Region member 2. schemaPartition: location of schema 3. dataPartition: location of
- * data
- */
-public class PartitionInfo implements SnapshotProcessor {
-
- private static final Logger LOGGER = LoggerFactory.getLogger(PartitionInfo.class);
-
- // Region read write lock
- private final ReentrantReadWriteLock regionReadWriteLock;
- private AtomicInteger nextRegionGroupId = new AtomicInteger(0);
- private final Map<TConsensusGroupId, TRegionReplicaSet> regionReplicaMap;
- // Map<TConsensusGroupId, allocatedSlotsNumber>
- private final Map<TConsensusGroupId, Long> regionSlotsCounter;
- // preDeleted TODO: Combine it with Partition.class
- private final Set<String> preDeletedStorageGroup = new CopyOnWriteArraySet<>();
- private final Set<TRegionReplicaSet> deletedRegionSet = new HashSet<>();
-
- // SchemaPartition read write lock
- private final ReentrantReadWriteLock schemaPartitionReadWriteLock;
- private final SchemaPartition schemaPartition;
-
- // DataPartition read write lock
- private final ReentrantReadWriteLock dataPartitionReadWriteLock;
- private final DataPartition dataPartition;
-
- private final String snapshotFileName = "partition_info.bin";
-
- public PartitionInfo() {
- this.regionReadWriteLock = new ReentrantReadWriteLock();
- this.regionReplicaMap = new HashMap<>();
- this.regionSlotsCounter = new HashMap<>();
-
- this.schemaPartitionReadWriteLock = new ReentrantReadWriteLock();
- this.schemaPartition =
- new SchemaPartition(
- ConfigNodeDescriptor.getInstance().getConf().getSeriesPartitionExecutorClass(),
- ConfigNodeDescriptor.getInstance().getConf().getSeriesPartitionSlotNum());
- this.schemaPartition.setSchemaPartitionMap(new HashMap<>());
-
- this.dataPartitionReadWriteLock = new ReentrantReadWriteLock();
- this.dataPartition =
- new DataPartition(
- ConfigNodeDescriptor.getInstance().getConf().getSeriesPartitionExecutorClass(),
- ConfigNodeDescriptor.getInstance().getConf().getSeriesPartitionSlotNum());
- this.dataPartition.setDataPartitionMap(new HashMap<>());
- }
-
- public int generateNextRegionGroupId() {
- return nextRegionGroupId.getAndIncrement();
- }
-
- @TestOnly
- public Integer getNextRegionGroupId() {
- return nextRegionGroupId.get();
- }
-
- /**
- * Persistence allocation result of new Regions
- *
- * @param req CreateRegionsPlan
- * @return SUCCESS_STATUS
- */
- public TSStatus createRegions(CreateRegionsReq req) {
- TSStatus result;
- regionReadWriteLock.writeLock().lock();
- try {
- int maxRegionId = Integer.MIN_VALUE;
-
- for (List<TRegionReplicaSet> regionReplicaSets : req.getRegionMap().values()) {
- for (TRegionReplicaSet regionReplicaSet : regionReplicaSets) {
- regionReplicaMap.put(regionReplicaSet.getRegionId(), regionReplicaSet);
- regionSlotsCounter.put(regionReplicaSet.getRegionId(), 0L);
- maxRegionId = Math.max(maxRegionId, regionReplicaSet.getRegionId().getId());
- }
- }
-
- if (nextRegionGroupId.get() < maxRegionId) {
- // In this case, at least one Region is created with the leader node,
- // so the nextRegionGroupID of the followers needs to be added
- nextRegionGroupId.getAndAdd(req.getRegionMap().size());
- }
-
- result = new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
- } finally {
- regionReadWriteLock.writeLock().unlock();
- }
- return result;
- }
-
- /**
- * Delete Regions
- *
- * @param req DeleteRegionsReq
- * @return SUCCESS_STATUS
- */
- public TSStatus deleteRegions(DeleteRegionsReq req) {
- TSStatus result;
- regionReadWriteLock.writeLock().lock();
- try {
- for (TConsensusGroupId consensusGroupId : req.getConsensusGroupIds()) {
- deletedRegionSet.add(regionReplicaMap.remove(consensusGroupId));
- regionSlotsCounter.remove(consensusGroupId);
- }
- result = new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
- } finally {
- regionReadWriteLock.writeLock().unlock();
- }
- return result;
- }
-
- /**
- * Delete StorageGroup
- *
- * @param req DeleteRegionsReq
- */
- public void deleteStorageGroup(DeleteStorageGroupReq req) {
- TStorageGroupSchema storageGroupSchema = req.getStorageGroup();
- List<TConsensusGroupId> dataRegionGroupIds = storageGroupSchema.getDataRegionGroupIds();
- List<TConsensusGroupId> schemaRegionGroupIds = storageGroupSchema.getSchemaRegionGroupIds();
- DeleteRegionsReq deleteRegionsReq = new DeleteRegionsReq();
- for (TConsensusGroupId schemaRegionGroupId : schemaRegionGroupIds) {
- deleteRegionsReq.addConsensusGroupId(schemaRegionGroupId);
- }
- for (TConsensusGroupId dataRegionId : dataRegionGroupIds) {
- deleteRegionsReq.addConsensusGroupId(dataRegionId);
- }
- deleteRegions(deleteRegionsReq);
- deleteDataPartitionMapByStorageGroup(storageGroupSchema.getName());
- deleteSchemaPartitionMapByStorageGroup(storageGroupSchema.getName());
- }
-
- /**
- * Get SchemaPartition
- *
- * @param req SchemaPartitionPlan with partitionSlotsMap
- * @return SchemaPartitionDataSet that contains only existing SchemaPartition
- */
- public DataSet getSchemaPartition(GetSchemaPartitionReq req) {
- SchemaPartitionResp schemaPartitionResp = new SchemaPartitionResp();
- schemaPartitionReadWriteLock.readLock().lock();
-
- try {
- schemaPartitionResp.setSchemaPartition(
- schemaPartition.getSchemaPartition(req.getPartitionSlotsMap(), preDeletedStorageGroup));
- } finally {
- schemaPartitionReadWriteLock.readLock().unlock();
- schemaPartitionResp.setStatus(new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode()));
- }
-
- return schemaPartitionResp;
- }
-
- /**
- * Create SchemaPartition
- *
- * @param req CreateSchemaPartitionPlan with SchemaPartition assigned result
- * @return TSStatusCode.SUCCESS_STATUS when creation successful
- */
- public TSStatus createSchemaPartition(CreateSchemaPartitionReq req) {
- schemaPartitionReadWriteLock.writeLock().lock();
- regionReadWriteLock.writeLock().lock();
-
- try {
- // Allocate SchemaPartition by CreateSchemaPartitionPlan
- Map<String, Map<TSeriesPartitionSlot, TRegionReplicaSet>> assignedResult =
- req.getAssignedSchemaPartition();
- assignedResult.forEach(
- (storageGroup, partitionSlots) ->
- partitionSlots.forEach(
- (seriesPartitionSlot, regionReplicaSet) -> {
- schemaPartition.createSchemaPartition(
- storageGroup, seriesPartitionSlot, regionReplicaSet);
- regionSlotsCounter.computeIfPresent(
- regionReplicaSet.getRegionId(), (consensusGroupId, count) -> (count + 1));
- }));
- } finally {
- regionReadWriteLock.writeLock().unlock();
- schemaPartitionReadWriteLock.writeLock().unlock();
- }
-
- return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
- }
-
- /**
- * Filter no assigned SchemaPartitionSlots
- *
- * @param partitionSlotsMap Map<StorageGroupName, List<TSeriesPartitionSlot>>
- * @return Map<StorageGroupName, List < TSeriesPartitionSlot>>, SchemaPartitionSlots that is not
- * assigned in partitionSlotsMap
- */
- public Map<String, List<TSeriesPartitionSlot>> filterNoAssignedSchemaPartitionSlots(
- Map<String, List<TSeriesPartitionSlot>> partitionSlotsMap) {
- Map<String, List<TSeriesPartitionSlot>> result;
- schemaPartitionReadWriteLock.readLock().lock();
- try {
- result = schemaPartition.filterNoAssignedSchemaPartitionSlot(partitionSlotsMap);
- } finally {
- schemaPartitionReadWriteLock.readLock().unlock();
- }
- return result;
- }
-
- /**
- * Get DataPartition
- *
- * @param req DataPartitionPlan with partitionSlotsMap
- * @return DataPartitionDataSet that contains only existing DataPartition
- */
- public DataSet getDataPartition(GetDataPartitionReq req) {
- DataPartitionResp dataPartitionResp = new DataPartitionResp();
- dataPartitionReadWriteLock.readLock().lock();
-
- try {
- dataPartitionResp.setDataPartition(
- dataPartition.getDataPartition(
- req.getPartitionSlotsMap(),
- ConfigNodeDescriptor.getInstance().getConf().getSeriesPartitionExecutorClass(),
- ConfigNodeDescriptor.getInstance().getConf().getSeriesPartitionSlotNum(),
- preDeletedStorageGroup));
- } finally {
- dataPartitionReadWriteLock.readLock().unlock();
- dataPartitionResp.setStatus(new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode()));
- }
-
- return dataPartitionResp;
- }
-
- /**
- * Create DataPartition
- *
- * @param req CreateDataPartitionPlan with DataPartition assigned result
- * @return TSStatusCode.SUCCESS_STATUS when creation successful
- */
- public TSStatus createDataPartition(CreateDataPartitionReq req) {
- dataPartitionReadWriteLock.writeLock().lock();
- regionReadWriteLock.writeLock().lock();
-
- try {
- // Allocate DataPartition by CreateDataPartitionPlan
- Map<String, Map<TSeriesPartitionSlot, Map<TTimePartitionSlot, List<TRegionReplicaSet>>>>
- assignedResult = req.getAssignedDataPartition();
- assignedResult.forEach(
- (storageGroup, seriesPartitionTimePartitionSlots) ->
- seriesPartitionTimePartitionSlots.forEach(
- ((seriesPartitionSlot, timePartitionSlotRegionReplicaSets) ->
- timePartitionSlotRegionReplicaSets.forEach(
- ((timePartitionSlot, regionReplicaSets) ->
- regionReplicaSets.forEach(
- regionReplicaSet -> {
- dataPartition.createDataPartition(
- storageGroup,
- seriesPartitionSlot,
- timePartitionSlot,
- regionReplicaSet);
- regionSlotsCounter.computeIfPresent(
- regionReplicaSet.getRegionId(),
- (consensusGroupId, count) -> (count + 1));
- }))))));
- } finally {
- regionReadWriteLock.writeLock().unlock();
- dataPartitionReadWriteLock.writeLock().unlock();
- }
-
- return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
- }
-
- /**
- * Filter no assigned DataPartitionSlots
- *
- * @param partitionSlotsMap Map<StorageGroupName, Map<TSeriesPartitionSlot,
- * List<TTimePartitionSlot>>>
- * @return Map<StorageGroupName, Map < TSeriesPartitionSlot, List < TTimePartitionSlot>>>,
- * DataPartitionSlots that is not assigned in partitionSlotsMap
- */
- public Map<String, Map<TSeriesPartitionSlot, List<TTimePartitionSlot>>>
- filterNoAssignedDataPartitionSlots(
- Map<String, Map<TSeriesPartitionSlot, List<TTimePartitionSlot>>> partitionSlotsMap) {
- Map<String, Map<TSeriesPartitionSlot, List<TTimePartitionSlot>>> result;
- dataPartitionReadWriteLock.readLock().lock();
- try {
- result = dataPartition.filterNoAssignedDataPartitionSlots(partitionSlotsMap);
- } finally {
- dataPartitionReadWriteLock.readLock().unlock();
- }
- return result;
- }
-
- /** Get RegionReplicaSet by the specific TConsensusGroupIds */
- public List<TRegionReplicaSet> getRegionReplicaSets(List<TConsensusGroupId> groupIds) {
- List<TRegionReplicaSet> result = new ArrayList<>();
- regionReadWriteLock.readLock().lock();
- try {
- for (TConsensusGroupId groupId : groupIds) {
- result.add(regionReplicaMap.get(groupId));
- }
- } finally {
- regionReadWriteLock.readLock().unlock();
- }
- return result;
- }
-
- /** Get all allocated RegionReplicaSets */
- public List<TRegionReplicaSet> getAllocatedRegions() {
- List<TRegionReplicaSet> result;
- regionReadWriteLock.readLock().lock();
- try {
- result = new ArrayList<>(regionReplicaMap.values());
- } finally {
- regionReadWriteLock.readLock().unlock();
- }
- return result;
- }
-
- /** @return A copy of regionReplicaMap */
- public Map<TConsensusGroupId, TRegionReplicaSet> getRegionReplicaMap() {
- Map<TConsensusGroupId, TRegionReplicaSet> result;
- regionReadWriteLock.readLock().lock();
- try {
- result = new HashMap<>(regionReplicaMap);
- } finally {
- regionReadWriteLock.readLock().unlock();
- }
- return result;
- }
-
- /** @return The specific Regions that sorted by the number of allocated slots */
- public List<Pair<Long, TConsensusGroupId>> getSortedRegionSlotsCounter(
- List<TConsensusGroupId> consensusGroupIds) {
- List<Pair<Long, TConsensusGroupId>> result = new ArrayList<>();
- regionReadWriteLock.readLock().lock();
- try {
- for (TConsensusGroupId consensusGroupId : consensusGroupIds) {
- result.add(new Pair<>(regionSlotsCounter.get(consensusGroupId), consensusGroupId));
- }
- result.sort(Comparator.comparingLong(Pair::getLeft));
- } finally {
- regionReadWriteLock.readLock().unlock();
- }
- return result;
- }
-
- private void deleteDataPartitionMapByStorageGroup(String storageGroup) {
- dataPartitionReadWriteLock.writeLock().lock();
- try {
- dataPartition.getDataPartitionMap().remove(storageGroup);
- preDeletedStorageGroup.remove(storageGroup);
- } finally {
- dataPartitionReadWriteLock.writeLock().unlock();
- }
- }
-
- private void deleteSchemaPartitionMapByStorageGroup(String storageGroup) {
- schemaPartitionReadWriteLock.writeLock().lock();
- try {
- schemaPartition.getSchemaPartitionMap().remove(storageGroup);
- preDeletedStorageGroup.remove(storageGroup);
- } finally {
- schemaPartitionReadWriteLock.writeLock().unlock();
- }
- }
-
- public TSStatus preDeleteStorageGroup(PreDeleteStorageGroupReq preDeleteStorageGroupReq) {
- final PreDeleteStorageGroupReq.PreDeleteType preDeleteType =
- preDeleteStorageGroupReq.getPreDeleteType();
- final String storageGroup = preDeleteStorageGroupReq.getStorageGroup();
- switch (preDeleteType) {
- case EXECUTE:
- preDeletedStorageGroup.add(storageGroup);
- break;
- case ROLLBACK:
- preDeletedStorageGroup.remove(storageGroup);
- break;
- }
- return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
- }
-
- public Set<TRegionReplicaSet> getDeletedRegionSet() {
- return deletedRegionSet;
- }
-
- public boolean processTakeSnapshot(File snapshotDir) throws TException, IOException {
-
- File snapshotFile = new File(snapshotDir, snapshotFileName);
- if (snapshotFile.exists() && snapshotFile.isFile()) {
- LOGGER.error(
- "Failed to take snapshot, because snapshot file [{}] is already exist.",
- snapshotFile.getAbsolutePath());
- return false;
- }
-
- // prevents temporary files from being damaged and cannot be deleted, which affects the next
- // snapshot operation.
- File tmpFile = new File(snapshotFile.getAbsolutePath() + "-" + UUID.randomUUID());
-
- lockAllRead();
- try (FileOutputStream fileOutputStream = new FileOutputStream(tmpFile);
- TIOStreamTransport tioStreamTransport = new TIOStreamTransport(fileOutputStream)) {
- TProtocol protocol = new TBinaryProtocol(tioStreamTransport);
-
- // serialize nextRegionGroupId
- ReadWriteIOUtils.write(nextRegionGroupId.get(), fileOutputStream);
- // serialize regionMap
- serializeRegionMap(fileOutputStream, protocol);
- // serialize deletedRegionSet
- serializeDeletedRegionSet(fileOutputStream, protocol);
- // serialize schemaPartition
- schemaPartition.serialize(fileOutputStream, protocol);
- // serialize dataPartition
- dataPartition.serialize(fileOutputStream, protocol);
- // write to file
- fileOutputStream.flush();
- fileOutputStream.close();
- // rename file
- return tmpFile.renameTo(snapshotFile);
- } finally {
- unlockAllRead();
- // with or without success, delete temporary files anyway
- for (int retry = 0; retry < 5; retry++) {
- if (!tmpFile.exists() || tmpFile.delete()) {
- break;
- } else {
- LOGGER.warn(
- "Can't delete temporary snapshot file: {}, retrying...", tmpFile.getAbsolutePath());
- }
- }
- }
- }
-
- public void processLoadSnapshot(File snapshotDir) throws TException, IOException {
-
- File snapshotFile = new File(snapshotDir, snapshotFileName);
- if (!snapshotFile.exists() || !snapshotFile.isFile()) {
- LOGGER.error(
- "Failed to load snapshot,snapshot file [{}] is not exist.",
- snapshotFile.getAbsolutePath());
- return;
- }
-
- // no operations are processed at this time
- lockAllWrite();
-
- try (FileInputStream fileInputStream = new FileInputStream(snapshotFile);
- TIOStreamTransport tioStreamTransport = new TIOStreamTransport(fileInputStream)) {
- TProtocol protocol = new TBinaryProtocol(tioStreamTransport);
- // before restoring a snapshot, clear all old data
- clear();
- // start to restore
- nextRegionGroupId.set(ReadWriteIOUtils.readInt(fileInputStream));
- deserializeRegionMap(fileInputStream, protocol);
- // deserialize deletedRegionSet
- deserializeDeletedRegionSet(fileInputStream, protocol);
- schemaPartition.deserialize(fileInputStream, protocol);
- dataPartition.deserialize(fileInputStream, protocol);
- } finally {
- unlockAllWrite();
- }
- }
-
- /** Get SchemaNodeManagementPartition through matched storageGroup */
- public DataSet getSchemaNodeManagementPartition(List<String> matchedStorageGroups) {
- SchemaNodeManagementResp schemaNodeManagementResp = new SchemaNodeManagementResp();
- schemaPartitionReadWriteLock.readLock().lock();
- try {
- schemaNodeManagementResp.setSchemaPartition(
- schemaPartition.getSchemaPartition(matchedStorageGroups));
- } finally {
- schemaPartitionReadWriteLock.readLock().unlock();
- schemaNodeManagementResp.setStatus(new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode()));
- }
- return schemaNodeManagementResp;
- }
-
- private void lockAllWrite() {
- regionReadWriteLock.writeLock().lock();
- schemaPartitionReadWriteLock.writeLock().lock();
- dataPartitionReadWriteLock.writeLock().lock();
- }
-
- private void unlockAllWrite() {
- regionReadWriteLock.writeLock().unlock();
- schemaPartitionReadWriteLock.writeLock().unlock();
- dataPartitionReadWriteLock.writeLock().unlock();
- }
-
- private void lockAllRead() {
- regionReadWriteLock.readLock().lock();
- schemaPartitionReadWriteLock.readLock().lock();
- dataPartitionReadWriteLock.readLock().lock();
- }
-
- private void unlockAllRead() {
- regionReadWriteLock.readLock().unlock();
- schemaPartitionReadWriteLock.readLock().unlock();
- dataPartitionReadWriteLock.readLock().unlock();
- }
-
- @TestOnly
- public DataPartition getDataPartition() {
- return dataPartition;
- }
-
- @TestOnly
- public SchemaPartition getSchemaPartition() {
- return schemaPartition;
- }
-
- @TestOnly
- public Map<TConsensusGroupId, Long> getRegionSlotsCounter() {
- return regionSlotsCounter;
- }
-
- private void serializeRegionMap(OutputStream outputStream, TProtocol protocol)
- throws TException, IOException {
- ReadWriteIOUtils.write(regionReplicaMap.size(), outputStream);
- for (TConsensusGroupId consensusGroupId : regionReplicaMap.keySet()) {
- consensusGroupId.write(protocol);
- regionReplicaMap.get(consensusGroupId).write(protocol);
- protocol.writeI64(regionSlotsCounter.get(consensusGroupId));
- }
- }
-
- private void deserializeRegionMap(InputStream inputStream, TProtocol protocol)
- throws TException, IOException {
- int size = ReadWriteIOUtils.readInt(inputStream);
- while (size > 0) {
- TConsensusGroupId tConsensusGroupId = new TConsensusGroupId();
- tConsensusGroupId.read(protocol);
- TRegionReplicaSet tRegionReplicaSet = new TRegionReplicaSet();
- tRegionReplicaSet.read(protocol);
- Long count = protocol.readI64();
-
- regionReplicaMap.put(tConsensusGroupId, tRegionReplicaSet);
- regionSlotsCounter.put(tConsensusGroupId, count);
- size--;
- }
- }
-
- private void serializeDeletedRegionSet(OutputStream outputStream, TProtocol protocol)
- throws TException, IOException {
- ReadWriteIOUtils.write(regionReplicaMap.size(), outputStream);
- for (TRegionReplicaSet regionReplicaSet : deletedRegionSet) {
- regionReplicaSet.write(protocol);
- }
- }
-
- private void deserializeDeletedRegionSet(InputStream inputStream, TProtocol protocol)
- throws TException, IOException {
- int size = ReadWriteIOUtils.readInt(inputStream);
- while (size > 0) {
- TRegionReplicaSet tRegionReplicaSet = new TRegionReplicaSet();
- tRegionReplicaSet.read(protocol);
- deletedRegionSet.add(tRegionReplicaSet);
- size--;
- }
- }
-
- public void clear() {
- nextRegionGroupId = new AtomicInteger(0);
- regionReplicaMap.clear();
- regionSlotsCounter.clear();
-
- if (schemaPartition.getSchemaPartitionMap() != null) {
- schemaPartition.getSchemaPartitionMap().clear();
- }
-
- if (dataPartition.getDataPartitionMap() != null) {
- dataPartition.getDataPartitionMap().clear();
- }
- }
-}
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/persistence/executor/ConfigRequestExecutor.java b/confignode/src/main/java/org/apache/iotdb/confignode/persistence/executor/ConfigRequestExecutor.java
index eaed02f226..f990370a80 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/persistence/executor/ConfigRequestExecutor.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/persistence/executor/ConfigRequestExecutor.java
@@ -36,7 +36,6 @@ import org.apache.iotdb.confignode.consensus.request.write.CreateFunctionReq;
import org.apache.iotdb.confignode.consensus.request.write.CreateRegionsReq;
import org.apache.iotdb.confignode.consensus.request.write.CreateSchemaPartitionReq;
import org.apache.iotdb.confignode.consensus.request.write.DeleteProcedureReq;
-import org.apache.iotdb.confignode.consensus.request.write.DeleteRegionsReq;
import org.apache.iotdb.confignode.consensus.request.write.DeleteStorageGroupReq;
import org.apache.iotdb.confignode.consensus.request.write.DropFunctionReq;
import org.apache.iotdb.confignode.consensus.request.write.PreDeleteStorageGroupReq;
@@ -52,9 +51,9 @@ import org.apache.iotdb.confignode.exception.physical.UnknownPhysicalPlanTypeExc
import org.apache.iotdb.confignode.persistence.AuthorInfo;
import org.apache.iotdb.confignode.persistence.ClusterSchemaInfo;
import org.apache.iotdb.confignode.persistence.NodeInfo;
-import org.apache.iotdb.confignode.persistence.PartitionInfo;
import org.apache.iotdb.confignode.persistence.ProcedureInfo;
import org.apache.iotdb.confignode.persistence.UDFInfo;
+import org.apache.iotdb.confignode.persistence.partition.PartitionInfo;
import org.apache.iotdb.consensus.common.DataSet;
import org.apache.iotdb.rpc.TSStatusCode;
import org.apache.iotdb.tsfile.utils.Pair;
@@ -142,7 +141,11 @@ public class ConfigRequestExecutor {
case RegisterDataNode:
return nodeInfo.registerDataNode((RegisterDataNodeReq) req);
case SetStorageGroup:
- return clusterSchemaInfo.setStorageGroup((SetStorageGroupReq) req);
+ TSStatus status = clusterSchemaInfo.setStorageGroup((SetStorageGroupReq) req);
+ if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ return status;
+ }
+ return partitionInfo.setStorageGroup((SetStorageGroupReq) req);
case DeleteStorageGroup:
partitionInfo.deleteStorageGroup((DeleteStorageGroupReq) req);
return clusterSchemaInfo.deleteStorageGroup((DeleteStorageGroupReq) req);
@@ -157,13 +160,7 @@ public class ConfigRequestExecutor {
case SetTimePartitionInterval:
return clusterSchemaInfo.setTimePartitionInterval((SetTimePartitionIntervalReq) req);
case CreateRegions:
- TSStatus status = clusterSchemaInfo.createRegions((CreateRegionsReq) req);
- if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
- return status;
- }
return partitionInfo.createRegions((CreateRegionsReq) req);
- case DeleteRegions:
- return partitionInfo.deleteRegions((DeleteRegionsReq) req);
case CreateSchemaPartition:
return partitionInfo.createSchemaPartition((CreateSchemaPartitionReq) req);
case CreateDataPartition:
@@ -241,7 +238,6 @@ public class ConfigRequestExecutor {
}
public void loadSnapshot(File latestSnapshotRootDir) {
-
if (!latestSnapshotRootDir.exists()) {
LOGGER.error(
"snapshot directory [{}] is not exist, can not load snapshot with this directory.",
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/persistence/partition/PartitionInfo.java b/confignode/src/main/java/org/apache/iotdb/confignode/persistence/partition/PartitionInfo.java
new file mode 100644
index 0000000000..73a72f803d
--- /dev/null
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/persistence/partition/PartitionInfo.java
@@ -0,0 +1,617 @@
+/*
+ * 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.persistence.partition;
+
+import org.apache.iotdb.common.rpc.thrift.TConsensusGroupId;
+import org.apache.iotdb.common.rpc.thrift.TConsensusGroupType;
+import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet;
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.common.rpc.thrift.TSeriesPartitionSlot;
+import org.apache.iotdb.common.rpc.thrift.TTimePartitionSlot;
+import org.apache.iotdb.commons.partition.DataPartition;
+import org.apache.iotdb.commons.partition.SchemaPartition;
+import org.apache.iotdb.commons.snapshot.SnapshotProcessor;
+import org.apache.iotdb.commons.utils.TestOnly;
+import org.apache.iotdb.confignode.conf.ConfigNodeDescriptor;
+import org.apache.iotdb.confignode.consensus.request.read.GetDataPartitionReq;
+import org.apache.iotdb.confignode.consensus.request.read.GetSchemaPartitionReq;
+import org.apache.iotdb.confignode.consensus.request.write.CreateDataPartitionReq;
+import org.apache.iotdb.confignode.consensus.request.write.CreateRegionsReq;
+import org.apache.iotdb.confignode.consensus.request.write.CreateSchemaPartitionReq;
+import org.apache.iotdb.confignode.consensus.request.write.DeleteStorageGroupReq;
+import org.apache.iotdb.confignode.consensus.request.write.PreDeleteStorageGroupReq;
+import org.apache.iotdb.confignode.consensus.request.write.SetStorageGroupReq;
+import org.apache.iotdb.confignode.consensus.response.DataPartitionResp;
+import org.apache.iotdb.confignode.consensus.response.SchemaNodeManagementResp;
+import org.apache.iotdb.confignode.consensus.response.SchemaPartitionResp;
+import org.apache.iotdb.confignode.exception.StorageGroupNotExistsException;
+import org.apache.iotdb.consensus.common.DataSet;
+import org.apache.iotdb.rpc.TSStatusCode;
+import org.apache.iotdb.tsfile.utils.Pair;
+import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
+
+import org.apache.thrift.TException;
+import org.apache.thrift.protocol.TBinaryProtocol;
+import org.apache.thrift.protocol.TProtocol;
+import org.apache.thrift.transport.TIOStreamTransport;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.UUID;
+import java.util.Vector;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CopyOnWriteArraySet;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * The PartitionInfo stores cluster PartitionTable. The PartitionTable including: 1. regionMap:
+ * location of Region member 2. schemaPartition: location of schema 3. dataPartition: location of
+ * data
+ */
+public class PartitionInfo implements SnapshotProcessor {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(PartitionInfo.class);
+
+ // For allocating Regions
+ private final AtomicInteger nextRegionGroupId;
+ // Map<StorageGroupName, StorageGroupPartitionInfo>
+ private final ConcurrentHashMap<String, StorageGroupPartitionTable> storageGroupPartitionTables;
+
+ // preDeleted TODO: Combine it with Partition.class
+ private final Set<String> preDeletedStorageGroup;
+ private final Set<TRegionReplicaSet> deletedRegionSet;
+
+ private final String snapshotFileName = "partition_info.bin";
+
+ public PartitionInfo() {
+ this.storageGroupPartitionTables = new ConcurrentHashMap<>();
+ this.nextRegionGroupId = new AtomicInteger(0);
+
+ // Ensure that the PartitionTables of the StorageGroups who've been logically deleted
+ // are unreadable and un-writable
+ this.preDeletedStorageGroup = new CopyOnWriteArraySet<>();
+ // For RegionCleaner
+ this.deletedRegionSet = Collections.synchronizedSet(new HashSet<>());
+ }
+
+ public int generateNextRegionGroupId() {
+ return nextRegionGroupId.getAndIncrement();
+ }
+
+ // ======================================================
+ // Consensus read/write interfaces
+ // ======================================================
+
+ /**
+ * Thread-safely create new StorageGroupPartitionInfo
+ *
+ * @param req SetStorageGroupReq
+ * @return SUCCESS_STATUS if the new StorageGroupPartitionInfo is created successfully.
+ */
+ public TSStatus setStorageGroup(SetStorageGroupReq req) {
+ storageGroupPartitionTables.put(req.getSchema().getName(), new StorageGroupPartitionTable());
+
+ LOGGER.info("Successfully set StorageGroup: {}", req.getSchema());
+
+ return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
+ }
+
+ /**
+ * Thread-safely cache allocation result of new Regions
+ *
+ * @param req CreateRegionsReq
+ * @return SUCCESS_STATUS
+ */
+ public TSStatus createRegions(CreateRegionsReq req) {
+ TSStatus result;
+ AtomicInteger maxRegionId = new AtomicInteger(Integer.MIN_VALUE);
+
+ req.getRegionMap()
+ .forEach(
+ (storageGroup, regionReplicaSets) -> {
+ storageGroupPartitionTables.get(storageGroup).createRegions(regionReplicaSets);
+ regionReplicaSets.forEach(
+ regionReplicaSet ->
+ maxRegionId.set(
+ Math.max(maxRegionId.get(), regionReplicaSet.getRegionId().getId())));
+ });
+
+ if (nextRegionGroupId.get() < maxRegionId.get()) {
+ // In this case, at least one Region is created by the leader ConfigNode,
+ // so the nextRegionGroupID of the followers needs to be added
+ nextRegionGroupId.getAndAdd(req.getRegionMap().size());
+ }
+
+ result = new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
+ return result;
+ }
+
+ /**
+ * Thread-safely pre-delete the specific StorageGroup
+ *
+ * @param preDeleteStorageGroupReq PreDeleteStorageGroupReq
+ * @return SUCCESS_STATUS
+ */
+ public TSStatus preDeleteStorageGroup(PreDeleteStorageGroupReq preDeleteStorageGroupReq) {
+ final PreDeleteStorageGroupReq.PreDeleteType preDeleteType =
+ preDeleteStorageGroupReq.getPreDeleteType();
+ final String storageGroup = preDeleteStorageGroupReq.getStorageGroup();
+ switch (preDeleteType) {
+ case EXECUTE:
+ preDeletedStorageGroup.add(storageGroup);
+ break;
+ case ROLLBACK:
+ preDeletedStorageGroup.remove(storageGroup);
+ break;
+ }
+ return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
+ }
+
+ /**
+ * Thread-safely delete StorageGroup
+ *
+ * @param req DeleteRegionsReq
+ */
+ public void deleteStorageGroup(DeleteStorageGroupReq req) {
+ // Cache RegionReplicaSets
+ synchronized (deletedRegionSet) {
+ deletedRegionSet.addAll(storageGroupPartitionTables.get(req.getName()).getAllReplicaSets());
+ }
+ // Clean the cache
+ storageGroupPartitionTables.remove(req.getName());
+ }
+
+ /** @return The Regions that should be deleted among the DataNodes */
+ public Set<TRegionReplicaSet> getDeletedRegionSet() {
+ synchronized (deletedRegionSet) {
+ return deletedRegionSet;
+ }
+ }
+
+ /**
+ * Thread-safely get SchemaPartition
+ *
+ * @param req SchemaPartitionPlan with partitionSlotsMap
+ * @return SchemaPartitionDataSet that contains only existing SchemaPartition
+ */
+ public DataSet getSchemaPartition(GetSchemaPartitionReq req) {
+ AtomicBoolean isAllPartitionsExist = new AtomicBoolean(true);
+ // TODO: Replace this map whit new SchemaPartition
+ Map<String, Map<TSeriesPartitionSlot, TRegionReplicaSet>> schemaPartition =
+ new ConcurrentHashMap<>();
+
+ if (req.getPartitionSlotsMap().size() == 0) {
+ // Return all SchemaPartitions when the queried PartitionSlots are empty
+ storageGroupPartitionTables.forEach(
+ (storageGroup, storageGroupPartitionTable) -> {
+ if (!preDeletedStorageGroup.contains(storageGroup)) {
+ schemaPartition.put(storageGroup, new ConcurrentHashMap<>());
+ storageGroupPartitionTable.getSchemaPartition(
+ new ArrayList<>(), schemaPartition.get(storageGroup));
+
+ if (schemaPartition.get(storageGroup).size() == 0) {
+ // Remove empty Map
+ schemaPartition.remove(storageGroup);
+ }
+ }
+ });
+ } else {
+ // Return the SchemaPartition for each StorageGroup
+ req.getPartitionSlotsMap()
+ .forEach(
+ (storageGroup, partitionSlots) -> {
+ if (storageGroupPartitionTables.containsKey(storageGroup)
+ && !preDeletedStorageGroup.contains(storageGroup)) {
+ schemaPartition.put(storageGroup, new ConcurrentHashMap<>());
+
+ if (!storageGroupPartitionTables
+ .get(storageGroup)
+ .getSchemaPartition(partitionSlots, schemaPartition.get(storageGroup))) {
+ isAllPartitionsExist.set(false);
+ }
+
+ if (schemaPartition.get(storageGroup).size() == 0) {
+ // Remove empty Map
+ schemaPartition.remove(storageGroup);
+ }
+ }
+ });
+ }
+
+ SchemaPartitionResp schemaPartitionResp = new SchemaPartitionResp();
+ schemaPartitionResp.setAllPartitionsExist(isAllPartitionsExist.get());
+ // Notice: SchemaPartition contains two redundant structural parameters currently.
+ schemaPartitionResp.setSchemaPartition(
+ new SchemaPartition(
+ schemaPartition,
+ ConfigNodeDescriptor.getInstance().getConf().getSeriesPartitionExecutorClass(),
+ ConfigNodeDescriptor.getInstance().getConf().getSeriesPartitionSlotNum()));
+ schemaPartitionResp.setStatus(new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode()));
+ return schemaPartitionResp;
+ }
+
+ /**
+ * Thread-safely get DataPartition
+ *
+ * @param req DataPartitionPlan with partitionSlotsMap
+ * @return DataPartitionDataSet that contains only existing DataPartition
+ */
+ public DataSet getDataPartition(GetDataPartitionReq req) {
+ AtomicBoolean isAllPartitionsExist = new AtomicBoolean(true);
+ // TODO: Replace this map whit new DataPartition
+ Map<String, Map<TSeriesPartitionSlot, Map<TTimePartitionSlot, List<TRegionReplicaSet>>>>
+ dataPartition = new ConcurrentHashMap<>();
+
+ req.getPartitionSlotsMap()
+ .forEach(
+ (storageGroup, partitionSlots) -> {
+ if (storageGroupPartitionTables.containsKey(storageGroup)
+ && !preDeletedStorageGroup.contains(storageGroup)) {
+ dataPartition.put(storageGroup, new ConcurrentHashMap<>());
+ if (!storageGroupPartitionTables
+ .get(storageGroup)
+ .getDataPartition(partitionSlots, dataPartition.get(storageGroup))) {
+ isAllPartitionsExist.set(false);
+ }
+
+ if (dataPartition.get(storageGroup).size() == 0) {
+ // Remove empty Map
+ dataPartition.remove(storageGroup);
+ }
+ }
+ });
+
+ DataPartitionResp dataPartitionResp = new DataPartitionResp();
+ dataPartitionResp.setAllPartitionsExist(isAllPartitionsExist.get());
+ // Notice: DataPartition contains two redundant structural parameters currently.
+ dataPartitionResp.setDataPartition(
+ new DataPartition(
+ dataPartition,
+ ConfigNodeDescriptor.getInstance().getConf().getSeriesPartitionExecutorClass(),
+ ConfigNodeDescriptor.getInstance().getConf().getSeriesPartitionSlotNum()));
+ dataPartitionResp.setStatus(new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode()));
+ return dataPartitionResp;
+ }
+
+ /**
+ * Create SchemaPartition
+ *
+ * @param req CreateSchemaPartitionPlan with SchemaPartition assigned result
+ * @return TSStatusCode.SUCCESS_STATUS
+ */
+ public TSStatus createSchemaPartition(CreateSchemaPartitionReq req) {
+ req.getAssignedSchemaPartition()
+ .forEach(
+ (storageGroup, schemaPartitionTable) -> {
+ if (storageGroupPartitionTables.containsKey(storageGroup)
+ && !preDeletedStorageGroup.contains(storageGroup)) {
+ storageGroupPartitionTables
+ .get(storageGroup)
+ .createSchemaPartition(schemaPartitionTable);
+ }
+ });
+
+ return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
+ }
+
+ /**
+ * Create DataPartition
+ *
+ * @param req CreateDataPartitionPlan with DataPartition assigned result
+ * @return TSStatusCode.SUCCESS_STATUS
+ */
+ public TSStatus createDataPartition(CreateDataPartitionReq req) {
+ req.getAssignedDataPartition()
+ .forEach(
+ (storageGroup, dataPartitionTable) -> {
+ if (storageGroupPartitionTables.containsKey(storageGroup)
+ && !preDeletedStorageGroup.contains(storageGroup)) {
+ storageGroupPartitionTables
+ .get(storageGroup)
+ .createDataPartition(dataPartitionTable);
+ }
+ });
+
+ return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
+ }
+
+ /** Get SchemaNodeManagementPartition through matched storageGroup */
+ public DataSet getSchemaNodeManagementPartition(List<String> matchedStorageGroups) {
+ SchemaNodeManagementResp schemaNodeManagementResp = new SchemaNodeManagementResp();
+ Map<String, Map<TSeriesPartitionSlot, TRegionReplicaSet>> schemaPartitionMap =
+ new ConcurrentHashMap<>();
+
+ matchedStorageGroups.forEach(
+ storageGroup -> {
+ schemaPartitionMap.put(storageGroup, new ConcurrentHashMap<>());
+ storageGroupPartitionTables
+ .get(storageGroup)
+ .getSchemaPartition(new ArrayList<>(), schemaPartitionMap.get(storageGroup));
+
+ if (schemaPartitionMap.get(storageGroup).size() == 0) {
+ // Remove empty Map
+ schemaPartitionMap.remove(storageGroup);
+ }
+ });
+
+ schemaNodeManagementResp.setSchemaPartition(
+ new SchemaPartition(
+ schemaPartitionMap,
+ ConfigNodeDescriptor.getInstance().getConf().getSeriesPartitionExecutorClass(),
+ ConfigNodeDescriptor.getInstance().getConf().getSeriesPartitionSlotNum()));
+ schemaNodeManagementResp.setStatus(new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode()));
+ return schemaNodeManagementResp;
+ }
+
+ // ======================================================
+ // Leader scheduling interfaces
+ // ======================================================
+
+ /**
+ * Only Leader use this interface. Filter unassigned SchemaPartitionSlots
+ *
+ * @param partitionSlotsMap Map<StorageGroupName, List<TSeriesPartitionSlot>>
+ * @return Map<StorageGroupName, List<TSeriesPartitionSlot>>, SchemaPartitionSlots that is not
+ * assigned in partitionSlotsMap
+ */
+ public Map<String, List<TSeriesPartitionSlot>> filterUnassignedSchemaPartitionSlots(
+ Map<String, List<TSeriesPartitionSlot>> partitionSlotsMap) {
+ Map<String, List<TSeriesPartitionSlot>> result = new ConcurrentHashMap<>();
+
+ partitionSlotsMap.forEach(
+ (storageGroup, partitionSlots) -> {
+ if (storageGroupPartitionTables.containsKey(storageGroup)
+ && !preDeletedStorageGroup.contains(storageGroup)) {
+ result.put(
+ storageGroup,
+ storageGroupPartitionTables
+ .get(storageGroup)
+ .filterUnassignedSchemaPartitionSlots(partitionSlots));
+ }
+ });
+
+ return result;
+ }
+
+ /**
+ * Only Leader use this interface. Filter unassigned SchemaPartitionSlots
+ *
+ * @param partitionSlotsMap Map<StorageGroupName, Map<TSeriesPartitionSlot,
+ * List<TTimePartitionSlot>>>
+ * @return Map<StorageGroupName, Map<TSeriesPartitionSlot, List<TTimePartitionSlot>>>,
+ * DataPartitionSlots that is not assigned in partitionSlotsMap
+ */
+ public Map<String, Map<TSeriesPartitionSlot, List<TTimePartitionSlot>>>
+ filterUnassignedDataPartitionSlots(
+ Map<String, Map<TSeriesPartitionSlot, List<TTimePartitionSlot>>> partitionSlotsMap) {
+ Map<String, Map<TSeriesPartitionSlot, List<TTimePartitionSlot>>> result =
+ new ConcurrentHashMap<>();
+
+ partitionSlotsMap.forEach(
+ (storageGroup, partitionSlots) -> {
+ if (storageGroupPartitionTables.containsKey(storageGroup)
+ && !preDeletedStorageGroup.contains(storageGroup)) {
+ result.put(
+ storageGroup,
+ storageGroupPartitionTables
+ .get(storageGroup)
+ .filterUnassignedDataPartitionSlots(partitionSlots));
+ }
+ });
+
+ return result;
+ }
+
+ /**
+ * Only leader use this interface.
+ *
+ * @return All Regions' RegionReplicaSet
+ */
+ public List<TRegionReplicaSet> getAllReplicaSets() {
+ List<TRegionReplicaSet> result = new Vector<>();
+ storageGroupPartitionTables
+ .values()
+ .forEach(
+ storageGroupPartitionTable ->
+ result.addAll(storageGroupPartitionTable.getAllReplicaSets()));
+ return result;
+ }
+
+ /**
+ * Only leader use this interface. Get the number of Regions currently owned by the specific
+ * StorageGroup
+ *
+ * @param storageGroup StorageGroupName
+ * @param type SchemaRegion or DataRegion
+ * @return Number of Regions currently owned by the specific StorageGroup
+ * @throws StorageGroupNotExistsException When the specific StorageGroup doesn't exist
+ */
+ public int getRegionCount(String storageGroup, TConsensusGroupType type)
+ throws StorageGroupNotExistsException {
+ if (!storageGroupPartitionTables.containsKey(storageGroup)
+ || preDeletedStorageGroup.contains(storageGroup)) {
+ throw new StorageGroupNotExistsException(storageGroup);
+ }
+
+ return storageGroupPartitionTables.get(storageGroup).getRegionCount(type);
+ }
+
+ /**
+ * Only leader use this interface. Contending the Region allocation particle
+ *
+ * @param storageGroup StorageGroupName
+ * @param type SchemaRegion or DataRegion
+ * @return True when successfully get the allocation particle, false otherwise
+ */
+ public boolean getRegionAllocationParticle(String storageGroup, TConsensusGroupType type) {
+ return storageGroupPartitionTables.get(storageGroup).getRegionAllocationParticle(type);
+ }
+
+ /**
+ * Only leader use this interface.
+ *
+ * @param storageGroup StorageGroupName
+ * @param type SchemaRegion or DataRegion
+ * @return The specific StorageGroup's Regions that sorted by the number of allocated slots
+ */
+ public List<Pair<Long, TConsensusGroupId>> getSortedRegionSlotsCounter(
+ String storageGroup, TConsensusGroupType type) {
+ return storageGroupPartitionTables.get(storageGroup).getSortedRegionSlotsCounter(type);
+ }
+
+ public boolean processTakeSnapshot(File snapshotDir) throws TException, IOException {
+
+ File snapshotFile = new File(snapshotDir, snapshotFileName);
+ if (snapshotFile.exists() && snapshotFile.isFile()) {
+ LOGGER.error(
+ "Failed to take snapshot, because snapshot file [{}] is already exist.",
+ snapshotFile.getAbsolutePath());
+ return false;
+ }
+
+ // prevents temporary files from being damaged and cannot be deleted, which affects the next
+ // snapshot operation.
+ File tmpFile = new File(snapshotFile.getAbsolutePath() + "-" + UUID.randomUUID());
+
+ // TODO: Lock PartitionInfo
+ try (FileOutputStream fileOutputStream = new FileOutputStream(tmpFile);
+ TIOStreamTransport tioStreamTransport = new TIOStreamTransport(fileOutputStream)) {
+ TProtocol protocol = new TBinaryProtocol(tioStreamTransport);
+
+ // serialize nextRegionGroupId
+ ReadWriteIOUtils.write(nextRegionGroupId.get(), fileOutputStream);
+ // serialize StorageGroupPartitionTable
+ ReadWriteIOUtils.write(storageGroupPartitionTables.size(), fileOutputStream);
+ for (Map.Entry<String, StorageGroupPartitionTable> storageGroupPartitionTableEntry :
+ storageGroupPartitionTables.entrySet()) {
+ ReadWriteIOUtils.write(storageGroupPartitionTableEntry.getKey(), fileOutputStream);
+ storageGroupPartitionTableEntry.getValue().serialize(fileOutputStream, protocol);
+ }
+ // serialize preDeletedStorageGroup
+ ReadWriteIOUtils.write(preDeletedStorageGroup.size(), fileOutputStream);
+ for (String storageGroup : preDeletedStorageGroup) {
+ ReadWriteIOUtils.write(storageGroup, fileOutputStream);
+ }
+ // serialize deletedRegionSet
+ ReadWriteIOUtils.write(deletedRegionSet.size(), fileOutputStream);
+ for (TRegionReplicaSet regionReplicaSet : deletedRegionSet) {
+ regionReplicaSet.write(protocol);
+ }
+
+ // write to file
+ fileOutputStream.flush();
+ fileOutputStream.close();
+ // rename file
+ return tmpFile.renameTo(snapshotFile);
+ } finally {
+ // with or without success, delete temporary files anyway
+ for (int retry = 0; retry < 5; retry++) {
+ if (!tmpFile.exists() || tmpFile.delete()) {
+ break;
+ } else {
+ LOGGER.warn(
+ "Can't delete temporary snapshot file: {}, retrying...", tmpFile.getAbsolutePath());
+ }
+ }
+ }
+ }
+
+ public void processLoadSnapshot(File snapshotDir) throws TException, IOException {
+
+ File snapshotFile = new File(snapshotDir, snapshotFileName);
+ if (!snapshotFile.exists() || !snapshotFile.isFile()) {
+ LOGGER.error(
+ "Failed to load snapshot,snapshot file [{}] is not exist.",
+ snapshotFile.getAbsolutePath());
+ return;
+ }
+
+ // TODO: Lock PartitionInfo
+ try (FileInputStream fileInputStream = new FileInputStream(snapshotFile);
+ TIOStreamTransport tioStreamTransport = new TIOStreamTransport(fileInputStream)) {
+ TProtocol protocol = new TBinaryProtocol(tioStreamTransport);
+ // before restoring a snapshot, clear all old data
+ clear();
+ // start to restore
+ nextRegionGroupId.set(ReadWriteIOUtils.readInt(fileInputStream));
+
+ // restore StorageGroupPartitionTable
+ int length = ReadWriteIOUtils.readInt(fileInputStream);
+ for (int i = 0; i < length; i++) {
+ String storageGroup = ReadWriteIOUtils.readString(fileInputStream);
+ StorageGroupPartitionTable storageGroupPartitionTable = new StorageGroupPartitionTable();
+ storageGroupPartitionTable.deserialize(fileInputStream, protocol);
+ storageGroupPartitionTables.put(storageGroup, storageGroupPartitionTable);
+ }
+ // restore preDeletedStorageGroup
+ length = ReadWriteIOUtils.readInt(fileInputStream);
+ for (int i = 0; i < length; i++) {
+ String storageGroup = ReadWriteIOUtils.readString(fileInputStream);
+ preDeletedStorageGroup.add(storageGroup);
+ }
+ // restore deletedRegionSet
+ length = ReadWriteIOUtils.readInt(fileInputStream);
+ for (int i = 0; i < length; i++) {
+ TRegionReplicaSet regionReplicaSet = new TRegionReplicaSet();
+ regionReplicaSet.read(protocol);
+ deletedRegionSet.add(regionReplicaSet);
+ }
+ }
+ }
+
+ public void clear() {
+ nextRegionGroupId.set(0);
+ storageGroupPartitionTables.clear();
+ preDeletedStorageGroup.clear();
+ deletedRegionSet.clear();
+ }
+
+ @TestOnly
+ public Set<String> getPreDeletedStorageGroup() {
+ return preDeletedStorageGroup;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+ PartitionInfo that = (PartitionInfo) o;
+ return storageGroupPartitionTables.equals(that.storageGroupPartitionTables)
+ && preDeletedStorageGroup.equals(that.preDeletedStorageGroup)
+ && deletedRegionSet.equals(that.deletedRegionSet);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(storageGroupPartitionTables, preDeletedStorageGroup, deletedRegionSet);
+ }
+}
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/persistence/partition/RegionGroup.java b/confignode/src/main/java/org/apache/iotdb/confignode/persistence/partition/RegionGroup.java
new file mode 100644
index 0000000000..cbe6a686f9
--- /dev/null
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/persistence/partition/RegionGroup.java
@@ -0,0 +1,97 @@
+/*
+ * 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.persistence.partition;
+
+import org.apache.iotdb.common.rpc.thrift.TConsensusGroupId;
+import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet;
+import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
+
+import org.apache.thrift.TException;
+import org.apache.thrift.protocol.TProtocol;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.Objects;
+import java.util.concurrent.atomic.AtomicLong;
+
+public class RegionGroup {
+
+ private final TConsensusGroupId id;
+ private final TRegionReplicaSet replicaSet;
+
+ // For DataRegion, each SeriesSlot * TimeSlot form a slot,
+ // for SchemaRegion, each SeriesSlot is a slot
+ private final AtomicLong slotCount;
+
+ public RegionGroup() {
+ this.id = new TConsensusGroupId();
+ this.replicaSet = new TRegionReplicaSet();
+ this.slotCount = new AtomicLong();
+ }
+
+ public RegionGroup(TRegionReplicaSet replicaSet) {
+ this.id = replicaSet.getRegionId();
+ this.replicaSet = replicaSet;
+ this.slotCount = new AtomicLong(0);
+ }
+
+ public TConsensusGroupId getId() {
+ return id;
+ }
+
+ public TRegionReplicaSet getReplicaSet() {
+ return replicaSet;
+ }
+
+ public void addCounter(long delta) {
+ slotCount.getAndAdd(delta);
+ }
+
+ public long getCounter() {
+ return slotCount.get();
+ }
+
+ public void serialize(OutputStream outputStream, TProtocol protocol)
+ throws IOException, TException {
+ id.write(protocol);
+ replicaSet.write(protocol);
+ ReadWriteIOUtils.write(slotCount.get(), outputStream);
+ }
+
+ public void deserialize(InputStream inputStream, TProtocol protocol)
+ throws IOException, TException {
+ id.read(protocol);
+ replicaSet.read(protocol);
+ slotCount.set(ReadWriteIOUtils.readLong(inputStream));
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+ RegionGroup that = (RegionGroup) o;
+ return id.equals(that.id) && replicaSet.equals(that.replicaSet);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(id, replicaSet);
+ }
+}
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/persistence/partition/StorageGroupPartitionTable.java b/confignode/src/main/java/org/apache/iotdb/confignode/persistence/partition/StorageGroupPartitionTable.java
new file mode 100644
index 0000000000..e857ffe2b9
--- /dev/null
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/persistence/partition/StorageGroupPartitionTable.java
@@ -0,0 +1,320 @@
+/*
+ * 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.persistence.partition;
+
+import org.apache.iotdb.common.rpc.thrift.TConsensusGroupId;
+import org.apache.iotdb.common.rpc.thrift.TConsensusGroupType;
+import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet;
+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.tsfile.utils.Pair;
+import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
+
+import org.apache.thrift.TException;
+import org.apache.thrift.protocol.TProtocol;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Vector;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+
+public class StorageGroupPartitionTable {
+
+ // Total number of SeriesPartitionSlots occupied by schema,
+ // determines whether a new Region needs to be created
+ private final AtomicInteger seriesPartitionSlotsCount;
+
+ // Region allocation particle
+ private final AtomicBoolean schemaRegionParticle;
+ private final AtomicBoolean dataRegionParticle;
+
+ // Region
+ private final Map<TConsensusGroupId, RegionGroup> regionInfoMap;
+ // SchemaPartition
+ private final SchemaPartitionTable schemaPartitionTable;
+ // DataPartition
+ private final DataPartitionTable dataPartitionTable;
+
+ public StorageGroupPartitionTable() {
+ this.seriesPartitionSlotsCount = new AtomicInteger(0);
+
+ this.schemaRegionParticle = new AtomicBoolean(true);
+ this.dataRegionParticle = new AtomicBoolean(true);
+ this.regionInfoMap = new ConcurrentHashMap<>();
+
+ this.schemaPartitionTable = new SchemaPartitionTable();
+ this.dataPartitionTable = new DataPartitionTable();
+ }
+
+ /**
+ * Cache allocation result of new Regions
+ *
+ * @param replicaSets List<TRegionReplicaSet>
+ */
+ public void createRegions(List<TRegionReplicaSet> replicaSets) {
+ replicaSets.forEach(
+ replicaSet -> regionInfoMap.put(replicaSet.getRegionId(), new RegionGroup(replicaSet)));
+ }
+
+ /** @return All Regions' RegionReplicaSet within one StorageGroup */
+ public List<TRegionReplicaSet> getAllReplicaSets() {
+ List<TRegionReplicaSet> result = new ArrayList<>();
+
+ for (RegionGroup regionGroup : regionInfoMap.values()) {
+ result.add(regionGroup.getReplicaSet());
+ }
+
+ return result;
+ }
+
+ /**
+ * Get the number of Regions currently owned by this StorageGroup
+ *
+ * @param type SchemaRegion or DataRegion
+ * @return The number of Regions currently owned by this StorageGroup
+ */
+ public int getRegionCount(TConsensusGroupType type) {
+ AtomicInteger result = new AtomicInteger(0);
+ regionInfoMap
+ .values()
+ .forEach(
+ regionGroup -> {
+ if (regionGroup.getId().getType().equals(type)) {
+ result.getAndIncrement();
+ }
+ });
+ return result.getAndIncrement();
+ }
+
+ /**
+ * Only leader use this interface. Contending the Region allocation particle
+ *
+ * @param type SchemaRegion or DataRegion
+ * @return True when successfully get the allocation particle, false otherwise
+ */
+ public boolean getRegionAllocationParticle(TConsensusGroupType type) {
+ switch (type) {
+ case SchemaRegion:
+ return schemaRegionParticle.getAndSet(false);
+ case DataRegion:
+ return dataRegionParticle.getAndSet(false);
+ default:
+ return false;
+ }
+ }
+
+ /**
+ * Thread-safely get SchemaPartition within the specific StorageGroup TODO: Remove mapping process
+ *
+ * @param partitionSlots SeriesPartitionSlots
+ * @param schemaPartition Where the results are stored
+ * @return True if all the SeriesPartitionSlots are matched, false otherwise
+ */
+ public boolean getSchemaPartition(
+ List<TSeriesPartitionSlot> partitionSlots,
+ Map<TSeriesPartitionSlot, TRegionReplicaSet> schemaPartition) {
+ // Get RegionIds
+ SchemaPartitionTable regionIds = new SchemaPartitionTable();
+ boolean result = schemaPartitionTable.getSchemaPartition(partitionSlots, regionIds);
+ // Map to RegionReplicaSets
+ regionIds
+ .getSchemaPartitionMap()
+ .forEach(
+ (seriesPartitionSlot, consensusGroupId) ->
+ schemaPartition.put(
+ seriesPartitionSlot, regionInfoMap.get(consensusGroupId).getReplicaSet()));
+ return result;
+ }
+
+ /**
+ * Thread-safely get DataPartition within the specific StorageGroup TODO: Remove mapping process
+ *
+ * @param partitionSlots SeriesPartitionSlots and TimePartitionSlots
+ * @param dataPartition Where the results are stored
+ * @return True if all the PartitionSlots are matched, false otherwise
+ */
+ public boolean getDataPartition(
+ Map<TSeriesPartitionSlot, List<TTimePartitionSlot>> partitionSlots,
+ Map<TSeriesPartitionSlot, Map<TTimePartitionSlot, List<TRegionReplicaSet>>> dataPartition) {
+ // Get RegionIds
+ DataPartitionTable regionIds = new DataPartitionTable();
+ boolean result = dataPartitionTable.getDataPartition(partitionSlots, regionIds);
+ // Map to RegionReplicaSets
+ regionIds
+ .getDataPartitionMap()
+ .forEach(
+ ((seriesPartitionSlot, seriesPartition) -> {
+ dataPartition.put(seriesPartitionSlot, new ConcurrentHashMap<>());
+ seriesPartition
+ .getSeriesPartitionMap()
+ .forEach(
+ ((timePartitionSlot, consensusGroupIds) -> {
+ dataPartition
+ .get(seriesPartitionSlot)
+ .put(timePartitionSlot, new Vector<>());
+ consensusGroupIds.forEach(
+ consensusGroupId ->
+ dataPartition
+ .get(seriesPartitionSlot)
+ .get(timePartitionSlot)
+ .add(regionInfoMap.get(consensusGroupId).getReplicaSet()));
+ }));
+ }));
+ return result;
+ }
+
+ /**
+ * Create SchemaPartition within the specific StorageGroup
+ *
+ * @param assignedSchemaPartition Assigned result
+ */
+ public void createSchemaPartition(SchemaPartitionTable assignedSchemaPartition) {
+ // Cache assigned result
+ Map<TConsensusGroupId, AtomicInteger> deltaMap =
+ schemaPartitionTable.createSchemaPartition(assignedSchemaPartition);
+
+ // Add counter
+ AtomicInteger total = new AtomicInteger(0);
+ deltaMap.forEach(
+ ((consensusGroupId, delta) -> {
+ total.getAndAdd(delta.get());
+ regionInfoMap.get(consensusGroupId).addCounter(delta.get());
+ }));
+ seriesPartitionSlotsCount.getAndAdd(total.get());
+ }
+
+ /**
+ * Create DataPartition within the specific StorageGroup
+ *
+ * @param assignedDataPartition Assigned result
+ */
+ public void createDataPartition(DataPartitionTable assignedDataPartition) {
+ // Cache assigned result
+ Map<TConsensusGroupId, AtomicInteger> deltaMap =
+ dataPartitionTable.createDataPartition(assignedDataPartition);
+
+ // Add counter
+ AtomicInteger total = new AtomicInteger(0);
+ deltaMap.forEach(
+ ((consensusGroupId, delta) -> {
+ total.getAndAdd(delta.get());
+ regionInfoMap.get(consensusGroupId).addCounter(delta.get());
+ }));
+ }
+
+ /**
+ * Only Leader use this interface. Filter unassigned SchemaPartitionSlots within the specific
+ * StorageGroup
+ *
+ * @param partitionSlots List<TSeriesPartitionSlot>
+ * @return Unassigned PartitionSlots
+ */
+ public List<TSeriesPartitionSlot> filterUnassignedSchemaPartitionSlots(
+ List<TSeriesPartitionSlot> partitionSlots) {
+ return schemaPartitionTable.filterUnassignedSchemaPartitionSlots(partitionSlots);
+ }
+
+ /**
+ * Only Leader use this interface. Filter unassigned DataPartitionSlots within the specific
+ * StorageGroup
+ *
+ * @param partitionSlots List<TSeriesPartitionSlot>
+ * @return Unassigned PartitionSlots
+ */
+ public Map<TSeriesPartitionSlot, List<TTimePartitionSlot>> filterUnassignedDataPartitionSlots(
+ Map<TSeriesPartitionSlot, List<TTimePartitionSlot>> partitionSlots) {
+ return dataPartitionTable.filterUnassignedDataPartitionSlots(partitionSlots);
+ }
+
+ /**
+ * Only leader use this interface.
+ *
+ * @param type SchemaRegion or DataRegion
+ * @return Regions that sorted by the number of allocated slots
+ */
+ public List<Pair<Long, TConsensusGroupId>> getSortedRegionSlotsCounter(TConsensusGroupType type) {
+ List<Pair<Long, TConsensusGroupId>> result = new Vector<>();
+
+ regionInfoMap.forEach(
+ (consensusGroupId, regionGroup) -> {
+ if (consensusGroupId.getType().equals(type)) {
+ result.add(new Pair<>(regionGroup.getCounter(), consensusGroupId));
+ }
+ });
+
+ result.sort(Comparator.comparingLong(Pair::getLeft));
+ return result;
+ }
+
+ public void serialize(OutputStream outputStream, TProtocol protocol)
+ throws IOException, TException {
+ ReadWriteIOUtils.write(seriesPartitionSlotsCount.get(), outputStream);
+
+ ReadWriteIOUtils.write(regionInfoMap.size(), outputStream);
+ for (Map.Entry<TConsensusGroupId, RegionGroup> regionInfoEntry : regionInfoMap.entrySet()) {
+ regionInfoEntry.getKey().write(protocol);
+ regionInfoEntry.getValue().serialize(outputStream, protocol);
+ }
+
+ schemaPartitionTable.serialize(outputStream, protocol);
+ dataPartitionTable.serialize(outputStream, protocol);
+ }
+
+ public void deserialize(InputStream inputStream, TProtocol protocol)
+ throws IOException, TException {
+ seriesPartitionSlotsCount.set(ReadWriteIOUtils.readInt(inputStream));
+
+ int length = ReadWriteIOUtils.readInt(inputStream);
+ for (int i = 0; i < length; i++) {
+ TConsensusGroupId consensusGroupId = new TConsensusGroupId();
+ consensusGroupId.read(protocol);
+ RegionGroup regionGroup = new RegionGroup();
+ regionGroup.deserialize(inputStream, protocol);
+ regionInfoMap.put(consensusGroupId, regionGroup);
+ }
+
+ schemaPartitionTable.deserialize(inputStream, protocol);
+ dataPartitionTable.deserialize(inputStream, protocol);
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+ StorageGroupPartitionTable that = (StorageGroupPartitionTable) o;
+ return regionInfoMap.equals(that.regionInfoMap)
+ && schemaPartitionTable.equals(that.schemaPartitionTable)
+ && dataPartitionTable.equals(that.dataPartitionTable);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(regionInfoMap, schemaPartitionTable, dataPartitionTable);
+ }
+}
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/procedure/env/ConfigNodeProcedureEnv.java b/confignode/src/main/java/org/apache/iotdb/confignode/procedure/env/ConfigNodeProcedureEnv.java
index 307355b6fa..bcddd90aa4 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/procedure/env/ConfigNodeProcedureEnv.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/procedure/env/ConfigNodeProcedureEnv.java
@@ -25,7 +25,6 @@ import org.apache.iotdb.confignode.client.SyncDataNodeClientPool;
import org.apache.iotdb.confignode.consensus.request.write.DeleteStorageGroupReq;
import org.apache.iotdb.confignode.consensus.request.write.PreDeleteStorageGroupReq;
import org.apache.iotdb.confignode.manager.ConfigManager;
-import org.apache.iotdb.confignode.rpc.thrift.TStorageGroupSchema;
import org.apache.iotdb.mpp.rpc.thrift.TInvalidateCacheReq;
import org.apache.iotdb.rpc.TSStatusCode;
@@ -64,13 +63,13 @@ public class ConfigNodeProcedureEnv {
}
/**
- * Delete config node information, includes (mTree, partitionInfo, regionMap)
+ * Delete ConfigNode cache, includes ClusterSchemaInfo and PartitionInfo
*
- * @param deleteSgSchema storage group name
+ * @param name storage group name
* @return tsStatus
*/
- public TSStatus deleteConfig(TStorageGroupSchema deleteSgSchema) {
- DeleteStorageGroupReq deleteStorageGroupReq = new DeleteStorageGroupReq(deleteSgSchema);
+ public TSStatus deleteConfig(String name) {
+ DeleteStorageGroupReq deleteStorageGroupReq = new DeleteStorageGroupReq(name);
return configManager.getClusterSchemaManager().deleteStorageGroup(deleteStorageGroupReq);
}
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/DeleteStorageGroupProcedure.java b/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/DeleteStorageGroupProcedure.java
index 8243bf8b53..3640fb0c46 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/DeleteStorageGroupProcedure.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/DeleteStorageGroupProcedure.java
@@ -93,7 +93,7 @@ public class DeleteStorageGroupProcedure
break;
case DELETE_CONFIG:
LOG.info("Delete config info of {}", deleteSgSchema.getName());
- TSStatus status = env.deleteConfig(deleteSgSchema);
+ TSStatus status = env.deleteConfig(deleteSgSchema.getName());
if (status.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
return Flow.NO_MORE_STATE;
} else if (getCycles() > retryThreshold) {
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCServiceProcessor.java b/confignode/src/main/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCServiceProcessor.java
index b589d5034f..11cbca37cf 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCServiceProcessor.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCServiceProcessor.java
@@ -91,7 +91,6 @@ import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.nio.ByteBuffer;
-import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.stream.Collectors;
@@ -182,10 +181,6 @@ public class ConfigNodeRPCServiceProcessor implements ConfigIService.Iface {
storageGroupSchema.setMaximumSchemaRegionCount(0);
storageGroupSchema.setMaximumDataRegionCount(0);
- // Initialize RegionGroupId List
- storageGroupSchema.setSchemaRegionGroupIds(new ArrayList<>());
- storageGroupSchema.setDataRegionGroupIds(new ArrayList<>());
-
SetStorageGroupReq setReq = new SetStorageGroupReq(storageGroupSchema);
TSStatus resp = configManager.setStorageGroup(setReq);
diff --git a/confignode/src/test/java/org/apache/iotdb/confignode/consensus/request/ConfigRequestSerDeTest.java b/confignode/src/test/java/org/apache/iotdb/confignode/consensus/request/ConfigRequestSerDeTest.java
index bc0f49c965..5d327923a6 100644
--- a/confignode/src/test/java/org/apache/iotdb/confignode/consensus/request/ConfigRequestSerDeTest.java
+++ b/confignode/src/test/java/org/apache/iotdb/confignode/consensus/request/ConfigRequestSerDeTest.java
@@ -29,6 +29,9 @@ import org.apache.iotdb.common.rpc.thrift.TSeriesPartitionSlot;
import org.apache.iotdb.common.rpc.thrift.TTimePartitionSlot;
import org.apache.iotdb.commons.auth.AuthException;
import org.apache.iotdb.commons.auth.entity.PrivilegeType;
+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.confignode.consensus.request.auth.AuthorReq;
import org.apache.iotdb.confignode.consensus.request.read.CountStorageGroupReq;
import org.apache.iotdb.confignode.consensus.request.read.GetDataNodeInfoReq;
@@ -55,12 +58,10 @@ import org.apache.iotdb.confignode.procedure.Procedure;
import org.apache.iotdb.confignode.procedure.impl.DeleteStorageGroupProcedure;
import org.apache.iotdb.confignode.rpc.thrift.TStorageGroupSchema;
-import org.junit.After;
import org.junit.Assert;
import org.junit.Test;
import java.io.IOException;
-import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
@@ -72,13 +73,6 @@ import java.util.Set;
public class ConfigRequestSerDeTest {
- private final ByteBuffer buffer = ByteBuffer.allocate(10240);
-
- @After
- public void cleanBuffer() {
- buffer.clear();
- }
-
@Test
public void RegisterDataNodeReqTest() throws IOException {
TDataNodeLocation dataNodeLocation = new TDataNodeLocation();
@@ -117,9 +111,7 @@ public class ConfigRequestSerDeTest {
.setTTL(Long.MAX_VALUE)
.setSchemaReplicationFactor(3)
.setDataReplicationFactor(3)
- .setTimePartitionInterval(604800)
- .setSchemaRegionGroupIds(new ArrayList<>())
- .setDataRegionGroupIds(new ArrayList<>()));
+ .setTimePartitionInterval(604800));
SetStorageGroupReq req1 =
(SetStorageGroupReq) ConfigRequest.Factory.create(req0.serializeToByteBuffer());
Assert.assertEquals(req0, req1);
@@ -127,13 +119,7 @@ public class ConfigRequestSerDeTest {
@Test
public void DeleteStorageGroupReqTest() throws IOException {
- TStorageGroupSchema storageGroupSchema = new TStorageGroupSchema();
- storageGroupSchema.setName("root.sg");
- storageGroupSchema.addToSchemaRegionGroupIds(
- new TConsensusGroupId(TConsensusGroupType.DataRegion, 1));
- storageGroupSchema.addToSchemaRegionGroupIds(
- new TConsensusGroupId(TConsensusGroupType.SchemaRegion, 2));
- DeleteStorageGroupReq req0 = new DeleteStorageGroupReq(storageGroupSchema);
+ DeleteStorageGroupReq req0 = new DeleteStorageGroupReq("root.sg");
DeleteStorageGroupReq req1 =
(DeleteStorageGroupReq) ConfigRequest.Factory.create(req0.serializeToByteBuffer());
Assert.assertEquals(req0, req1);
@@ -220,8 +206,8 @@ public class ConfigRequestSerDeTest {
@Test
public void DeleteRegionsPlanTest() throws IOException {
DeleteRegionsReq req0 = new DeleteRegionsReq();
- req0.addConsensusGroupId(new TConsensusGroupId(TConsensusGroupType.SchemaRegion, 0));
- req0.addConsensusGroupId(new TConsensusGroupId(TConsensusGroupType.DataRegion, 1));
+ req0.addDeleteRegion("sg", new TConsensusGroupId(TConsensusGroupType.SchemaRegion, 0));
+ req0.addDeleteRegion("sg", new TConsensusGroupId(TConsensusGroupType.DataRegion, 1));
DeleteRegionsReq req1 =
(DeleteRegionsReq) ConfigRequest.Factory.create(req0.serializeToByteBuffer());
@@ -240,14 +226,12 @@ public class ConfigRequestSerDeTest {
String storageGroup = "root.sg0";
TSeriesPartitionSlot seriesPartitionSlot = new TSeriesPartitionSlot(10);
- TRegionReplicaSet regionReplicaSet = new TRegionReplicaSet();
- regionReplicaSet.setRegionId(new TConsensusGroupId(TConsensusGroupType.SchemaRegion, 0));
- regionReplicaSet.setDataNodeLocations(Collections.singletonList(dataNodeLocation));
+ TConsensusGroupId consensusGroupId = new TConsensusGroupId(TConsensusGroupType.SchemaRegion, 0);
- Map<String, Map<TSeriesPartitionSlot, TRegionReplicaSet>> assignedSchemaPartition =
- new HashMap<>();
- assignedSchemaPartition.put(storageGroup, new HashMap<>());
- assignedSchemaPartition.get(storageGroup).put(seriesPartitionSlot, regionReplicaSet);
+ Map<String, SchemaPartitionTable> assignedSchemaPartition = new HashMap<>();
+ Map<TSeriesPartitionSlot, TConsensusGroupId> schemaPartitionMap = new HashMap<>();
+ schemaPartitionMap.put(seriesPartitionSlot, consensusGroupId);
+ assignedSchemaPartition.put(storageGroup, new SchemaPartitionTable(schemaPartitionMap));
CreateSchemaPartitionReq req0 = new CreateSchemaPartitionReq();
req0.setAssignedSchemaPartition(assignedSchemaPartition);
@@ -303,19 +287,15 @@ public class ConfigRequestSerDeTest {
regionReplicaSet.setRegionId(new TConsensusGroupId(TConsensusGroupType.DataRegion, 0));
regionReplicaSet.setDataNodeLocations(Collections.singletonList(dataNodeLocation));
- Map<String, Map<TSeriesPartitionSlot, Map<TTimePartitionSlot, List<TRegionReplicaSet>>>>
- assignedDataPartition = new HashMap<>();
- assignedDataPartition.put(storageGroup, new HashMap<>());
- assignedDataPartition.get(storageGroup).put(seriesPartitionSlot, new HashMap<>());
- assignedDataPartition
- .get(storageGroup)
- .get(seriesPartitionSlot)
- .put(timePartitionSlot, new ArrayList<>());
- assignedDataPartition
- .get(storageGroup)
- .get(seriesPartitionSlot)
- .get(timePartitionSlot)
- .add(regionReplicaSet);
+ Map<String, DataPartitionTable> assignedDataPartition = new HashMap<>();
+ Map<TSeriesPartitionSlot, SeriesPartitionTable> dataPartitionMap = new HashMap<>();
+ Map<TTimePartitionSlot, List<TConsensusGroupId>> seriesPartitionMap = new HashMap<>();
+
+ seriesPartitionMap.put(
+ timePartitionSlot,
+ Collections.singletonList(new TConsensusGroupId(TConsensusGroupType.DataRegion, 0)));
+ dataPartitionMap.put(seriesPartitionSlot, new SeriesPartitionTable(seriesPartitionMap));
+ assignedDataPartition.put(storageGroup, new DataPartitionTable(dataPartitionMap));
CreateDataPartitionReq req0 = new CreateDataPartitionReq();
req0.setAssignedDataPartition(assignedDataPartition);
@@ -377,13 +357,11 @@ public class ConfigRequestSerDeTest {
ConfigRequestType.CreateUser, "thulab", "", "passwd", "", new HashSet<>(), "");
req1 = (AuthorReq) ConfigRequest.Factory.create(req0.serializeToByteBuffer());
Assert.assertEquals(req0, req1);
- cleanBuffer();
// create role
req0 = new AuthorReq(ConfigRequestType.CreateRole, "", "admin", "", "", new HashSet<>(), "");
req1 = (AuthorReq) ConfigRequest.Factory.create(req0.serializeToByteBuffer());
Assert.assertEquals(req0, req1);
- cleanBuffer();
// alter user
req0 =
@@ -391,14 +369,12 @@ public class ConfigRequestSerDeTest {
ConfigRequestType.UpdateUser, "tempuser", "", "", "newpwd", new HashSet<>(), "");
req1 = (AuthorReq) ConfigRequest.Factory.create(req0.serializeToByteBuffer());
Assert.assertEquals(req0, req1);
- cleanBuffer();
// grant user
req0 =
new AuthorReq(ConfigRequestType.GrantUser, "tempuser", "", "", "", permissions, "root.ln");
req1 = (AuthorReq) ConfigRequest.Factory.create(req0.serializeToByteBuffer());
Assert.assertEquals(req0, req1);
- cleanBuffer();
// grant role
req0 =
@@ -412,27 +388,23 @@ public class ConfigRequestSerDeTest {
"root.ln");
req1 = (AuthorReq) ConfigRequest.Factory.create(req0.serializeToByteBuffer());
Assert.assertEquals(req0, req1);
- cleanBuffer();
// grant role to user
req0 = new AuthorReq(ConfigRequestType.GrantRole, "", "temprole", "", "", new HashSet<>(), "");
req1 = (AuthorReq) ConfigRequest.Factory.create(req0.serializeToByteBuffer());
Assert.assertEquals(req0, req1);
- cleanBuffer();
// revoke user
req0 =
new AuthorReq(ConfigRequestType.RevokeUser, "tempuser", "", "", "", permissions, "root.ln");
req1 = (AuthorReq) ConfigRequest.Factory.create(req0.serializeToByteBuffer());
Assert.assertEquals(req0, req1);
- cleanBuffer();
// revoke role
req0 =
new AuthorReq(ConfigRequestType.RevokeRole, "", "temprole", "", "", permissions, "root.ln");
req1 = (AuthorReq) ConfigRequest.Factory.create(req0.serializeToByteBuffer());
Assert.assertEquals(req0, req1);
- cleanBuffer();
// revoke role from user
req0 =
@@ -446,67 +418,56 @@ public class ConfigRequestSerDeTest {
"");
req1 = (AuthorReq) ConfigRequest.Factory.create(req0.serializeToByteBuffer());
Assert.assertEquals(req0, req1);
- cleanBuffer();
// drop user
req0 = new AuthorReq(ConfigRequestType.DropUser, "xiaoming", "", "", "", new HashSet<>(), "");
req1 = (AuthorReq) ConfigRequest.Factory.create(req0.serializeToByteBuffer());
Assert.assertEquals(req0, req1);
- cleanBuffer();
// drop role
req0 = new AuthorReq(ConfigRequestType.DropRole, "", "admin", "", "", new HashSet<>(), "");
req1 = (AuthorReq) ConfigRequest.Factory.create(req0.serializeToByteBuffer());
Assert.assertEquals(req0, req1);
- cleanBuffer();
// list user
req0 = new AuthorReq(ConfigRequestType.ListUser, "", "", "", "", new HashSet<>(), "");
req1 = (AuthorReq) ConfigRequest.Factory.create(req0.serializeToByteBuffer());
Assert.assertEquals(req0, req1);
- cleanBuffer();
// list role
req0 = new AuthorReq(ConfigRequestType.ListRole, "", "", "", "", new HashSet<>(), "");
req1 = (AuthorReq) ConfigRequest.Factory.create(req0.serializeToByteBuffer());
Assert.assertEquals(req0, req1);
- cleanBuffer();
// list privileges user
req0 = new AuthorReq(ConfigRequestType.ListUserPrivilege, "", "", "", "", new HashSet<>(), "");
req1 = (AuthorReq) ConfigRequest.Factory.create(req0.serializeToByteBuffer());
Assert.assertEquals(req0, req1);
- cleanBuffer();
// list privileges role
req0 = new AuthorReq(ConfigRequestType.ListRolePrivilege, "", "", "", "", new HashSet<>(), "");
req1 = (AuthorReq) ConfigRequest.Factory.create(req0.serializeToByteBuffer());
Assert.assertEquals(req0, req1);
- cleanBuffer();
// list user privileges
req0 = new AuthorReq(ConfigRequestType.ListUserPrivilege, "", "", "", "", new HashSet<>(), "");
req1 = (AuthorReq) ConfigRequest.Factory.create(req0.serializeToByteBuffer());
Assert.assertEquals(req0, req1);
- cleanBuffer();
// list role privileges
req0 = new AuthorReq(ConfigRequestType.ListRolePrivilege, "", "", "", "", new HashSet<>(), "");
req1 = (AuthorReq) ConfigRequest.Factory.create(req0.serializeToByteBuffer());
Assert.assertEquals(req0, req1);
- cleanBuffer();
// list all role of user
req0 = new AuthorReq(ConfigRequestType.ListUserRoles, "", "", "", "", new HashSet<>(), "");
req1 = (AuthorReq) ConfigRequest.Factory.create(req0.serializeToByteBuffer());
Assert.assertEquals(req0, req1);
- cleanBuffer();
// list all user of role
req0 = new AuthorReq(ConfigRequestType.ListRoleUsers, "", "", "", "", new HashSet<>(), "");
req1 = (AuthorReq) ConfigRequest.Factory.create(req0.serializeToByteBuffer());
Assert.assertEquals(req0, req1);
- cleanBuffer();
}
@Test
@@ -525,10 +486,6 @@ public class ConfigRequestSerDeTest {
DeleteStorageGroupProcedure procedure = new DeleteStorageGroupProcedure();
TStorageGroupSchema storageGroupSchema = new TStorageGroupSchema();
storageGroupSchema.setName("root.sg");
- storageGroupSchema.setSchemaRegionGroupIds(
- Collections.singletonList(new TConsensusGroupId(TConsensusGroupType.DataRegion, 0)));
- storageGroupSchema.setDataRegionGroupIds(
- Collections.singletonList(new TConsensusGroupId(TConsensusGroupType.SchemaRegion, 1)));
procedure.setDeleteSgSchema(storageGroupSchema);
UpdateProcedureReq updateProcedureReq = new UpdateProcedureReq();
updateProcedureReq.setProcedure(procedure);
@@ -545,14 +502,6 @@ public class ConfigRequestSerDeTest {
DeleteStorageGroupProcedure deleteStorageGroupProcedure = new DeleteStorageGroupProcedure();
TStorageGroupSchema tStorageGroupSchema = new TStorageGroupSchema();
tStorageGroupSchema.setName("root.sg");
- List<TConsensusGroupId> dataRegionIds = new ArrayList<>();
- List<TConsensusGroupId> schemaRegionIds = new ArrayList<>();
- TConsensusGroupId dataRegionId = new TConsensusGroupId(TConsensusGroupType.DataRegion, 1);
- dataRegionIds.add(dataRegionId);
- TConsensusGroupId schemaRegionId = new TConsensusGroupId(TConsensusGroupType.SchemaRegion, 2);
- schemaRegionIds.add(schemaRegionId);
- tStorageGroupSchema.setDataRegionGroupIds(dataRegionIds);
- tStorageGroupSchema.setSchemaRegionGroupIds(schemaRegionIds);
deleteStorageGroupProcedure.setDeleteSgSchema(tStorageGroupSchema);
req0.setProcedure(deleteStorageGroupProcedure);
UpdateProcedureReq req1 =
diff --git a/confignode/src/test/java/org/apache/iotdb/confignode/manager/hash/DeviceGroupHashExecutorManualTest.java b/confignode/src/test/java/org/apache/iotdb/confignode/manager/hash/DeviceGroupHashExecutorManualTest.java
index 47675f349e..0cb02123fb 100644
--- a/confignode/src/test/java/org/apache/iotdb/confignode/manager/hash/DeviceGroupHashExecutorManualTest.java
+++ b/confignode/src/test/java/org/apache/iotdb/confignode/manager/hash/DeviceGroupHashExecutorManualTest.java
@@ -20,7 +20,7 @@ package org.apache.iotdb.confignode.manager.hash;
import org.apache.iotdb.confignode.manager.ConfigManager;
import org.apache.iotdb.confignode.manager.PartitionManager;
-import org.apache.iotdb.confignode.persistence.PartitionInfo;
+import org.apache.iotdb.confignode.persistence.partition.PartitionInfo;
import java.io.IOException;
import java.util.ArrayList;
diff --git a/confignode/src/test/java/org/apache/iotdb/confignode/persistence/ClusterSchemaInfoTest.java b/confignode/src/test/java/org/apache/iotdb/confignode/persistence/ClusterSchemaInfoTest.java
index 514a38664c..388bda0998 100644
--- a/confignode/src/test/java/org/apache/iotdb/confignode/persistence/ClusterSchemaInfoTest.java
+++ b/confignode/src/test/java/org/apache/iotdb/confignode/persistence/ClusterSchemaInfoTest.java
@@ -95,17 +95,5 @@ public class ClusterSchemaInfoTest {
Map<String, TStorageGroupSchema> reloadResult =
clusterSchemaInfo.getMatchedStorageGroupSchemas(getStorageGroupReq).getSchemaMap();
Assert.assertEquals(testMap, reloadResult);
-
- Assert.assertEquals(
- clusterSchemaInfo.getSchemaRegionParticles().size(), storageGroupPathList.size());
- storageGroupPathList.forEach(
- storageGroup ->
- Assert.assertTrue(clusterSchemaInfo.getSchemaRegionParticles().contains(storageGroup)));
-
- Assert.assertEquals(
- clusterSchemaInfo.getDataRegionParticles().size(), storageGroupPathList.size());
- storageGroupPathList.forEach(
- storageGroup ->
- Assert.assertTrue(clusterSchemaInfo.getDataRegionParticles().contains(storageGroup)));
}
}
diff --git a/confignode/src/test/java/org/apache/iotdb/confignode/persistence/PartitionInfoTest.java b/confignode/src/test/java/org/apache/iotdb/confignode/persistence/PartitionInfoTest.java
index 49409b0a24..1eb83c8ad4 100644
--- a/confignode/src/test/java/org/apache/iotdb/confignode/persistence/PartitionInfoTest.java
+++ b/confignode/src/test/java/org/apache/iotdb/confignode/persistence/PartitionInfoTest.java
@@ -26,9 +26,15 @@ import org.apache.iotdb.common.rpc.thrift.TEndPoint;
import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet;
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.confignode.consensus.request.write.CreateDataPartitionReq;
import org.apache.iotdb.confignode.consensus.request.write.CreateRegionsReq;
import org.apache.iotdb.confignode.consensus.request.write.CreateSchemaPartitionReq;
+import org.apache.iotdb.confignode.consensus.request.write.SetStorageGroupReq;
+import org.apache.iotdb.confignode.persistence.partition.PartitionInfo;
+import org.apache.iotdb.confignode.rpc.thrift.TStorageGroupSchema;
import org.apache.commons.io.FileUtils;
import org.apache.thrift.TException;
@@ -88,6 +94,9 @@ public class PartitionInfoTest {
partitionInfo.generateNextRegionGroupId();
+ // Set StorageGroup
+ partitionInfo.setStorageGroup(new SetStorageGroupReq(new TStorageGroupSchema("root.test")));
+
// Create a SchemaRegion
CreateRegionsReq createRegionsReq = new CreateRegionsReq();
TRegionReplicaSet schemaRegionReplicaSet =
@@ -124,48 +133,18 @@ public class PartitionInfoTest {
testFlag.DataPartition.getFlag(), TConsensusGroupType.DataRegion));
partitionInfo.createDataPartition(createDataPartitionReq);
- Map<String, Map<TSeriesPartitionSlot, Map<TTimePartitionSlot, List<TRegionReplicaSet>>>>
- dataMap_before = partitionInfo.getDataPartition().getDataPartitionMap();
- int nextId = partitionInfo.getNextRegionGroupId();
+ partitionInfo.getPreDeletedStorageGroup().add("A");
+ partitionInfo.getPreDeletedStorageGroup().add("B");
+ partitionInfo.getPreDeletedStorageGroup().add("C");
- Map<TConsensusGroupId, Long> counter_before = partitionInfo.getRegionSlotsCounter();
partitionInfo.getDeletedRegionSet().add(dataRegionReplicaSet);
partitionInfo.getDeletedRegionSet().add(schemaRegionReplicaSet);
+
partitionInfo.processTakeSnapshot(snapshotDir);
- partitionInfo.clear();
- partitionInfo.processLoadSnapshot(snapshotDir);
-
- Assert.assertEquals(nextId, (int) partitionInfo.getNextRegionGroupId());
-
- // Check SchemaRegion
- List<TRegionReplicaSet> reloadTRegionReplicaSet =
- partitionInfo.getRegionReplicaSets(
- Collections.singletonList(
- generateTConsensusGroupId(
- testFlag.SchemaPartition.getFlag(), TConsensusGroupType.SchemaRegion)));
- Assert.assertEquals(1, reloadTRegionReplicaSet.size());
- Assert.assertEquals(schemaRegionReplicaSet, reloadTRegionReplicaSet.get(0));
-
- // Check DataRegion
- reloadTRegionReplicaSet =
- partitionInfo.getRegionReplicaSets(
- Collections.singletonList(
- generateTConsensusGroupId(
- testFlag.DataPartition.getFlag(), TConsensusGroupType.DataRegion)));
- Assert.assertEquals(1, reloadTRegionReplicaSet.size());
- Assert.assertEquals(dataRegionReplicaSet, reloadTRegionReplicaSet.get(0));
-
- Assert.assertEquals(
- createSchemaPartitionReq.getAssignedSchemaPartition(),
- partitionInfo.getSchemaPartition().getSchemaPartitionMap());
-
- Assert.assertEquals(2, partitionInfo.getRegionSlotsCounter().size());
- Assert.assertEquals(counter_before, partitionInfo.getRegionSlotsCounter());
-
- Assert.assertEquals(dataMap_before, partitionInfo.getDataPartition().getDataPartitionMap());
- Assert.assertEquals(2, partitionInfo.getDeletedRegionSet().size());
- Assert.assertTrue(partitionInfo.getDeletedRegionSet().contains(dataRegionReplicaSet));
- Assert.assertTrue(partitionInfo.getDeletedRegionSet().contains(schemaRegionReplicaSet));
+
+ PartitionInfo partitionInfo1 = new PartitionInfo();
+ partitionInfo1.processLoadSnapshot(snapshotDir);
+ Assert.assertEquals(partitionInfo, partitionInfo1);
}
private TRegionReplicaSet generateTRegionReplicaSet(
@@ -192,13 +171,10 @@ public class PartitionInfoTest {
int startFlag, TConsensusGroupId tConsensusGroupId) {
CreateSchemaPartitionReq createSchemaPartitionReq = new CreateSchemaPartitionReq();
// Map<StorageGroup, Map<TSeriesPartitionSlot, TSchemaRegionPlaceInfo>>
- Map<String, Map<TSeriesPartitionSlot, TRegionReplicaSet>> assignedSchemaPartition =
- new HashMap<>();
- Map<TSeriesPartitionSlot, TRegionReplicaSet> relationInfo = new HashMap<>();
- relationInfo.put(
- new TSeriesPartitionSlot(startFlag),
- generateTRegionReplicaSet(startFlag, tConsensusGroupId));
- assignedSchemaPartition.put("root.test.sg", relationInfo);
+ Map<String, SchemaPartitionTable> assignedSchemaPartition = new HashMap<>();
+ Map<TSeriesPartitionSlot, TConsensusGroupId> relationInfo = new HashMap<>();
+ relationInfo.put(new TSeriesPartitionSlot(startFlag), tConsensusGroupId);
+ assignedSchemaPartition.put("root.test", new SchemaPartitionTable(relationInfo));
createSchemaPartitionReq.setAssignedSchemaPartition(assignedSchemaPartition);
return createSchemaPartitionReq;
}
@@ -208,29 +184,23 @@ public class PartitionInfoTest {
startFlag = startFlag / 10;
CreateDataPartitionReq createSchemaPartitionReq = new CreateDataPartitionReq();
// Map<StorageGroup, Map<TSeriesPartitionSlot, Map<TTimePartitionSlot, List<TRegionMessage>>>>
- Map<String, Map<TSeriesPartitionSlot, Map<TTimePartitionSlot, List<TRegionReplicaSet>>>>
- dataPartitionMap = new HashMap<>();
-
- Map<TTimePartitionSlot, List<TRegionReplicaSet>> relationInfo = new HashMap<>();
+ Map<String, DataPartitionTable> dataPartitionMap = new HashMap<>();
- List<TRegionReplicaSet> tRegionReplicaSets = new ArrayList<>();
+ Map<TTimePartitionSlot, List<TConsensusGroupId>> relationInfo = new HashMap<>();
for (int i = 0; i <= startFlag; i++) {
- for (int j = 0; j <= startFlag; j++) {
- tRegionReplicaSets.add(generateTRegionReplicaSet(j + startFlag, tConsensusGroupId));
- }
relationInfo.put(
- new TTimePartitionSlot((System.currentTimeMillis() / 1000) + i), tRegionReplicaSets);
+ new TTimePartitionSlot((System.currentTimeMillis() / 1000) + i),
+ Collections.singletonList(tConsensusGroupId));
}
- Map<TSeriesPartitionSlot, Map<TTimePartitionSlot, List<TRegionReplicaSet>>> slotInfo =
- new HashMap<>();
+ Map<TSeriesPartitionSlot, SeriesPartitionTable> slotInfo = new HashMap<>();
for (int i = 0; i <= startFlag; i++) {
- slotInfo.put(new TSeriesPartitionSlot(startFlag + i), relationInfo);
+ slotInfo.put(new TSeriesPartitionSlot(startFlag + i), new SeriesPartitionTable(relationInfo));
}
- dataPartitionMap.put("root.test.data.sg", slotInfo);
+ dataPartitionMap.put("root.test", new DataPartitionTable(slotInfo));
createSchemaPartitionReq.setAssignedDataPartition(dataPartitionMap);
return createSchemaPartitionReq;
}
diff --git a/integration/checkstyle.xml b/integration/checkstyle.xml
index 3d0e9efbda..6070cd880d 100644
--- a/integration/checkstyle.xml
+++ b/integration/checkstyle.xml
@@ -25,193 +25,7 @@
<property name="severity" value="error"/>
<property name="fileExtensions" value="java"/>
<module name="BeforeExecutionExclusionFileFilter">
- <property name="fileNamePattern"
- value="(^.*([\\/]src[\\/]main|[\\/]src[\\/]test[\\/]java[\\/]org[\\/]apache[\\/]iotdb[\\/]session)[\\/].*$)|
- |^.*IoTDBSchemaTemplateIT\.java$|
- |^.*IoTDBSeriesReaderIT\.java$|
- |^.*IoTDBCompactionIT\.java$|
- |^.*IoTDBAggregationByLevelIT\.java$|
- |^.*IoTDBUDFNestAggregationIT\.java$|
- |^.*IoTDBAggregationDeleteIT\.java$|
- |^.*IoTDBUserDefinedAggregationFunctionIT\.java$|
- |^.*IoTDBAggregationIT\.java$|
- |^.*IoTDBAggregationSmallDataIT\.java$|
- |^.*IoTDBAggregationLargeDataIT\.java$|
- |^.*IoTDBPartialInsertionIT\.java$|
- |^.*IoTDBInsertWithoutTimeIT\.java$|
- |^.*IoTDBUDFManagementIT\.java$|
- |^.*IoTDBMultiSeriesIT\.java$|
- |^.*IoTDBDisableAlignIT\.java$|
- |^.*IoTDBManageTsFileResourceIT\.java$|
- |^.*IoTDBInsertMultiRowIT\.java$|
- |^.*IoTDBQueryWithComplexValueFilterIT\.java$|
- |^.*IoTDBUDTFHybridQueryIT\.java$|
- |^.*IoTDBQueryMemoryControlIT\.java$|
- |^.*IoTDBClearCacheIT\.java$|
- |^.*IoTDBTracingIT\.java$|
- |^.*IoTDBLoadExternalTsFileWithTimePartitionIT\.java$|
- |^.*IoTDBKillQueryIT\.java$|
- |^.*IoTDBCompactionWithIDTableIT\.java$|
- |^.*IoTDBMultiStatementsIT\.java$|
- |^.*IoTDBArithmeticIT\.java$|
- |^.*IoTDBQueryWithIDTableIT\.java$|
- |^.*IoTDBDaemonIT\.java$|
- |^.*IoTDBEncodingIT\.java$|
- |^.*IoTDBNestedQueryIT\.java$|
- |^.*IoTDBSelectSchemaIT\.java$|
- |^.*IoTDBSortedShowTimeseriesIT\.java$|
- |^.*IoTDBFillIT\.java$|
- |^.*IoTDBSelectCompareExpressionIT\.java$|
- |^.*IoTDBFilePathUtilsIT\.java$|
- |^.*IoTDBTtlIT\.java$|
- |^.*IoTDBUnseqOverlappedPageIT\.java$|
- |^.*IoTDBVersionIT\.java$|
- |^.*IoTDBCreateTimeseriesIT\.java$|
- |^.*IoTDBCompressTypeIT\.java$|
- |^.*IoTDBTagAlterIT\.java$|
- |^.*IoTDBQueryDemoIT\.java$|
- |^.*IoTDBNumberPathIT\.java$|
- |^.*IoTDBSettleIT\.java$|
- |^.*IoTDBOverlappedPageIT\.java$|
- |^.*IoTDBDatetimeFormatIT\.java$|
- |^.*IoTDBInIT\.java$|
- |^.*IoTDBMetadataFetchIT\.java$|
- |^.*IoTDBFloatPrecisionIT\.java$|
- |^.*IoTDBRecoverIT\.java$|
- |^.*IoTDBFuzzyQueryIT\.java$|
- |^.*IoTDBAliasIT\.java$|
- |^.*IoTDBUDFWindowQueryIT\.java$|
- |^.*IoTDBAuthorizationIT\.java$|
- |^.*IoTDBTriggerManagementIT\.java$|
- |^.*IoTDBRecoverUnclosedIT\.java$|
- |^.*IoTDBCreateAlignedTimeseriesIT\.java$|
- |^.*IoTDBSameMeasurementsDifferentTypesIT\.java$|
- |^.*IoTDBRpcCompressionIT\.java$|
- |^.*IoTDBAlignByDeviceIT\.java$|
- |^.*IoTDBLastQueryWithoutLastCacheWithDeletion2IT\.java$|
- |^.*IoTDBGroupByQueryWithoutValueFilterIT\.java$|
- |^.*IoTDBInsertAlignedValuesIT\.java$|
- |^.*IoTDBAggregationWithoutValueFilterWithDeletion2IT\.java$|
- |^.*IoTDBAggregationWithoutValueFilterIT\.java$|
- |^.*IoTDBRawQueryWithoutValueFilterWithDeletion2IT\.java$|
- |^.*IoTDBLoadExternalAlignedTsFileIT\.java$|
- |^.*IoTDBLastQueryWithDeletionIT\.java$|
- |^.*IoTDBGroupByQueryWithValueFilterWithDeletion2IT\.java$|
- |^.*IoTDBGroupBySlidingWindowQueryWithoutValueFilterIT\.java$|
- |^.*IoTDBRawQueryWithValueFilterIT\.java$|
- |^.*IoTDBLastQuery2IT\.java$|
- |^.*IoTDBLastQueryWithDeletion2IT\.java$|
- |^.*IoTDBAggregationWithValueFilterIT\.java$|
- |^.*IoTDBRawQueryWithoutValueFilter3IT\.java$|
- |^.*IoTDBLastQueryIT\.java$|
- |^.*IoTDBGroupByFillQueryBigDataIT\.java$|
- |^.*IoTDBGroupByQueryWithValueFilterWithDeletionIT\.java$|
- |^.*IoTDBAlignByDeviceIT\.java$|
- |^.*IoTDBAggregationWithValueFilterWithDeletion2IT\.java$|
- |^.*IoTDBGroupByQueryWithoutValueFilterWithDeletion2IT\.java$|
- |^.*IoTDBAggregationWithoutValueFilter2IT\.java$|
- |^.*IoTDBAggregationWithValueFilter2IT\.java$|
- |^.*IoTDBRawQueryWithValueFilterWithDeletion2IT\.java$|
- |^.*IoTDBGroupByFillQueryIT\.java$|
- |^.*IoTDBDeleteTimeseriesIT\.java$|
- |^.*IoTDBDeletionIT\.java$|
- |^.*IoTDBRawQueryWithoutValueFilterWithDeletionIT\.java$|
- |^.*IoTDBLastQueryWithoutLastCache2IT\.java$|
- |^.*IoTDBRawQueryWithoutValueFilterIT\.java$|
- |^.*IoTDBLastQueryWithoutLastCacheIT\.java$|
- |^.*IoTDBGroupByQueryWithoutValueFilter2IT\.java$|
- |^.*IoTDBGroupByQueryWithValueFilter2IT\.java$|
- |^.*IoTDBGroupByQueryWithoutValueFilterWithDeletionIT\.java$|
- |^.*IoTDBAggregationWithValueFilterWithDeletionIT\.java$|
- |^.*IoTDBRawQueryWithValueFilterWithDeletionIT\.java$|
- |^.*IoTDBAggregationGroupByLevelIT\.java$|
- |^.*IoTDBAlignByDevice2IT\.java$|
- |^.*IoTDBRawQueryWithValueFilter2IT\.java$|
- |^.*IoTDBGroupBySlidingWindowQueryWithValueFilterIT\.java$|
- |^.*IoTDBRawQueryWithoutValueFilter2IT\.java$|
- |^.*AlignedWriteUtil\.java$|
- |^.*IoTDBLastQueryWithoutLastCacheWithDeletionIT\.java$|
- |^.*IoTDBGroupByQueryWithValueFilterIT\.java$|
- |^.*IoTDBAggregationWithoutValueFilterWithDeletionIT\.java$|
- |^.*IoTDBMultiDeviceIT\.java$|
- |^.*IoTDBQueryVersionAdaptionIT\.java$|
- |^.*IoTDBSyntaxConventionVersionAdaptionIT\.java$|
- |^.*IoTDBDDLVersionAdaptionIT\.java$|
- |^.*IoTDBDeletionVersionAdaptionIT\.java$|
- |^.*IoTDBCreateStorageGroupIT\.java$|
- |^.*StandaloneEnv\.java$|
- |^.*StandaloneEnvConfig\.java$|
- |^.*IoTDBCompleteIT\.java$|
- |^.*IOTDBGroupByInnerIntervalIT\.java$|
- |^.*IoTDBGroupByFillWithRangeIT\.java$|
- |^.*IoTDBGroupByUnseqIT\.java$|
- |^.*IOTDBGroupByIT\.java$|
- |^.*IoTDBGroupByFillIT\.java$|
- |^.*IoTDBGroupByMonthFillIT\.java$|
- |^.*IoTDBGroupByFillMixPathsIT\.java$|
- |^.*IoTDBGroupByMonthIT\.java$|
- |^.*IoTDBResultSetIT\.java$|
- |^.*IoTDBSimpleQueryIT\.java$|
- |^.*IoTDBDeleteStorageGroupIT\.java$|
- |^.*IOTDBInsertIT\.java$|
- |^.*IoTDBTagIT\.java$|
- |^.*IoTDBDeleteTimeseriesIT\.java$|
- |^.*IoTDBDeletionIT\.java$|
- |^.*IoTDBExecuteBatchIT\.java$|
- |^.*IoTDBMaxTimeQueryIT\.java$|
- |^.*IoTDBSensorUpdateIT\.java$|
- |^.*IoTDBWithoutNullAllFilterIT\.java$|
- |^.*IoTDBWithoutNullAnyFilterIT\.java$|
- |^.*IoTDBRemovePartitionIT\.java$|
- |^.*IoTDBUDTFAlignByTimeQueryIT\.java$|
- |^.*IoTDBAutoCreateSchemaIT\.java$|
- |^.*IoTDBQueryTimeoutIT\.java$|
- |^.*IoTDBLoadExternalTsfileIT\.java$|
- |^.*IoTDBSessionTimeoutIT\.java$|
- |^.*IoTDBAsIT\.java$|
- |^.*IoTDBPathNumOverLimitIT\.java$|
- |^.*IoTDBFlushQueryMergeIT\.java$|
- |^.*IoTDBSelectIntoIT\.java$|
- |^.*IoTDBMultiOverlappedChunkInUnseqIT\.java$|
- |^.*IoTDBInsertWithQueryIT\.java$|
- |^.*IoTDBTimeZoneIT\.java$|
- |^.*IoTDBTriggerExecutionIT\.java$|
- |^.*IoTDBWithoutAllNullIT\.java$|
- |^.*IoTDBLimitSlimitIT\.java$|
- |^.*IoTDBLargeDataIT\.java$|
- |^.*IoTDBResultMetadataIT\.java$|
- |^.*IoTDBWithoutAnyNullIT\.java$|
- |^.*IoTDBQuotedPathIT\.java$|
- |^.*TransportClientMock\.java$|
- |^.*IoTDBSyncReceiverLoaderIT\.java$|
- |^.*IoTDBSyncSenderIT\.java$|
- |^.*IoTDBSyncReceiverIT\.java$|
- |^.*IoTDBSnapshotIT\.java$|
- |^.*IoTDBTriggerForwardIT\.java$|
- |^.*TransportHandlerMock\.java$|
- |^.*IoTDBSyntaxConventionStringLiteralIT\.java|
- |^.*IoTDBSyncReceiverCollectorIT\.java$|
- |^.*SyncTestUtil\.java$|
- |^.*IoTDBUDTFBuiltinFunctionIT\.java$|
- |^.*IoTDBInsertNaNIT\.java$|
- |^.*IoTDBLastIT\.java$|
- |^.*IoTDBNewTsFileCompactionIT\.java$|
- |^.*IoTDBSequenceDataQueryIT\.java$|
- |^.*IoTDBTimePartitionIT\.java$|
- |^.*IoTDBCloseIT\.java$|
- |^.*IoTDBQueryWithRecreatedTimeseriesIT\.java$|
- |^.*IoTDBEngineTimeGeneratorIT\.java$|
- |^.*IoTDBCheckConfigIT\.java$|
- |^.*IoTDBMultiOverlappedPageIT\.java$|
- |^.*IoTDBAlignedTimeSeriesCompactionIT\.java$|
- |^.*IoTDBRestartIT\.java$|
- |^.*IoTDBSyntaxConventionIT\.java$|
- |^.*IoTDBSyntaxConventionIdentifierIT\.java$|
- |^.*IoTDBInsertNullIT\.java$|
- |^.*IoTDBUDTFNonAlignQueryIT\.java$|
- |^.*IoTDBSetSystemReadOnlyWritableIT\.java$|
- |^.*IoTDBContinuousQueryIT\.java$|
- |^.*IoTDBSizeTieredCompactionIT\.java$"/>
+ <property name="fileNamePattern" value="(^.*([\\/]src[\\/]main|[\\/]src[\\/]test[\\/]java[\\/]org[\\/]apache[\\/]iotdb[\\/]session)[\\/].*$)| |^.*IoTDBSchemaTemplateIT\.java$| |^.*IoTDBSeriesReaderIT\.java$| |^.*IoTDBCompactionIT\.java$| |^.*IoTDBAggregationByLevelIT\.java$| |^.*IoTDBUDFNestAggregationIT\.java$| |^.*IoTDBAggregationDeleteIT\.java$| |^.*IoTDBUserDefinedA [...]
</module>
<!-- <module name="RegexpOnFilename">-->
<!-- <property name="fileNamePattern" value="^.*IT\.java$"/>-->
diff --git a/node-commons/src/main/java/org/apache/iotdb/commons/partition/DataPartition.java b/node-commons/src/main/java/org/apache/iotdb/commons/partition/DataPartition.java
index 4f343dab68..2e9804d9df 100644
--- a/node-commons/src/main/java/org/apache/iotdb/commons/partition/DataPartition.java
+++ b/node-commons/src/main/java/org/apache/iotdb/commons/partition/DataPartition.java
@@ -21,24 +21,15 @@ package org.apache.iotdb.commons.partition;
import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet;
import org.apache.iotdb.common.rpc.thrift.TSeriesPartitionSlot;
import org.apache.iotdb.common.rpc.thrift.TTimePartitionSlot;
-import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
-import org.apache.thrift.TException;
-import org.apache.thrift.protocol.TProtocol;
-
-import java.io.IOException;
-import java.io.InputStream;
-import java.io.OutputStream;
import java.util.ArrayList;
import java.util.Collection;
-import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
-import java.util.Map.Entry;
-import java.util.Set;
import java.util.stream.Collectors;
+// TODO: Remove this class
public class DataPartition extends Partition {
// Map<StorageGroup, Map<TSeriesPartitionSlot, Map<TTimePartitionSlot, List<TRegionMessage>>>>
@@ -126,194 +117,6 @@ public class DataPartition extends Partition {
return null;
}
- /* Interfaces for ConfigNode */
-
- /**
- * Get DataPartition by partitionSlotsMap
- *
- * @param partitionSlotsMap Map<StorageGroupName, Map<SeriesPartitionSlot,
- * List<TimePartitionSlot>>>
- * @return Subset of current DataPartition, including Map<StorageGroupName,
- * Map<SeriesPartitionSlot, Map<TimePartitionSlot, List<RegionReplicaSet>>>>
- */
- public DataPartition getDataPartition(
- Map<String, Map<TSeriesPartitionSlot, List<TTimePartitionSlot>>> partitionSlotsMap,
- String seriesSlotExecutorName,
- int seriesPartitionSlotNum,
- Set<String> preDeletedStorageGroup) {
- Map<String, Map<TSeriesPartitionSlot, Map<TTimePartitionSlot, List<TRegionReplicaSet>>>>
- result = new HashMap<>();
-
- for (String storageGroupName : partitionSlotsMap.keySet()) {
- // Compare StorageGroupName
- if (dataPartitionMap.containsKey(storageGroupName)
- && !preDeletedStorageGroup.contains(storageGroupName)) {
- Map<TSeriesPartitionSlot, Map<TTimePartitionSlot, List<TRegionReplicaSet>>>
- seriesTimePartitionSlotMap = dataPartitionMap.get(storageGroupName);
- for (TSeriesPartitionSlot seriesPartitionSlot :
- partitionSlotsMap.get(storageGroupName).keySet()) {
- // Compare SeriesPartitionSlot
- if (seriesTimePartitionSlotMap.containsKey(seriesPartitionSlot)) {
- Map<TTimePartitionSlot, List<TRegionReplicaSet>> timePartitionSlotMap =
- seriesTimePartitionSlotMap.get(seriesPartitionSlot);
- if (partitionSlotsMap.get(storageGroupName).get(seriesPartitionSlot).size() == 0) {
- result
- .computeIfAbsent(storageGroupName, key -> new HashMap<>())
- .computeIfAbsent(seriesPartitionSlot, key -> new HashMap<>())
- .putAll(new HashMap<>(timePartitionSlotMap));
- } else {
- for (TTimePartitionSlot timePartitionSlot :
- partitionSlotsMap.get(storageGroupName).get(seriesPartitionSlot)) {
- // Compare TimePartitionSlot
- if (timePartitionSlotMap.containsKey(timePartitionSlot)) {
- result
- .computeIfAbsent(storageGroupName, key -> new HashMap<>())
- .computeIfAbsent(seriesPartitionSlot, key -> new HashMap<>())
- .put(
- timePartitionSlot,
- new ArrayList<>(timePartitionSlotMap.get(timePartitionSlot)));
- }
- }
- }
- }
- }
- }
- }
-
- return new DataPartition(result, seriesSlotExecutorName, seriesPartitionSlotNum);
- }
-
- /**
- * Filter out unassigned PartitionSlots
- *
- * @param partitionSlotsMap Map<StorageGroupName, Map<SeriesPartitionSlot,
- * List<TimePartitionSlot>>>
- * @return Map<StorageGroupName, Map < SeriesPartitionSlot, List < TimePartitionSlot>>>,
- * unassigned PartitionSlots
- */
- public Map<String, Map<TSeriesPartitionSlot, List<TTimePartitionSlot>>>
- filterNoAssignedDataPartitionSlots(
- Map<String, Map<TSeriesPartitionSlot, List<TTimePartitionSlot>>> partitionSlotsMap) {
- Map<String, Map<TSeriesPartitionSlot, List<TTimePartitionSlot>>> result = new HashMap<>();
-
- for (String storageGroupName : partitionSlotsMap.keySet()) {
- // Compare StorageGroupName
- if (dataPartitionMap.containsKey(storageGroupName)) {
- Map<TSeriesPartitionSlot, Map<TTimePartitionSlot, List<TRegionReplicaSet>>>
- seriesTimePartitionSlotMap = dataPartitionMap.get(storageGroupName);
- for (TSeriesPartitionSlot seriesPartitionSlot :
- partitionSlotsMap.get(storageGroupName).keySet()) {
- // Compare SeriesPartitionSlot
- if (seriesTimePartitionSlotMap.containsKey(seriesPartitionSlot)) {
- Map<TTimePartitionSlot, List<TRegionReplicaSet>> timePartitionSlotMap =
- seriesTimePartitionSlotMap.get(seriesPartitionSlot);
- for (TTimePartitionSlot timePartitionSlot :
- partitionSlotsMap.get(storageGroupName).get(seriesPartitionSlot)) {
- // Compare TimePartitionSlot
- if (!timePartitionSlotMap.containsKey(timePartitionSlot)) {
- result
- .computeIfAbsent(storageGroupName, key -> new HashMap<>())
- .computeIfAbsent(seriesPartitionSlot, key -> new ArrayList<>())
- .add(timePartitionSlot);
- }
- }
- } else {
- // Clone all if SeriesPartitionSlot not assigned
- result
- .computeIfAbsent(storageGroupName, key -> new HashMap<>())
- .put(
- seriesPartitionSlot,
- new ArrayList<>(
- partitionSlotsMap.get(storageGroupName).get(seriesPartitionSlot)));
- }
- }
- } else {
- // Clone all if StorageGroupName not assigned
- result.put(storageGroupName, new HashMap<>(partitionSlotsMap.get(storageGroupName)));
- }
- }
-
- return result;
- }
-
- /** Create a DataPartition by ConfigNode */
- public void createDataPartition(
- String storageGroup,
- TSeriesPartitionSlot seriesPartitionSlot,
- TTimePartitionSlot timePartitionSlot,
- TRegionReplicaSet regionReplicaSet) {
- dataPartitionMap
- .computeIfAbsent(storageGroup, key -> new HashMap<>())
- .computeIfAbsent(seriesPartitionSlot, key -> new HashMap<>())
- .put(timePartitionSlot, Collections.singletonList(regionReplicaSet));
- }
-
- public void serialize(OutputStream outputStream, TProtocol protocol)
- throws IOException, TException {
- // Map<StorageGroup, Map<TSeriesPartitionSlot, Map<TTimePartitionSlot, List<TRegionMessage>>>>
- ReadWriteIOUtils.write(dataPartitionMap.size(), outputStream);
- for (Entry<String, Map<TSeriesPartitionSlot, Map<TTimePartitionSlot, List<TRegionReplicaSet>>>>
- entry : dataPartitionMap.entrySet()) {
- ReadWriteIOUtils.write(entry.getKey(), outputStream);
- ReadWriteIOUtils.write(entry.getValue().size(), outputStream);
- for (Entry<TSeriesPartitionSlot, Map<TTimePartitionSlot, List<TRegionReplicaSet>>>
- seriesPartitionSlotMapEntry : entry.getValue().entrySet()) {
- seriesPartitionSlotMapEntry.getKey().write(protocol);
- ReadWriteIOUtils.write(seriesPartitionSlotMapEntry.getValue().size(), outputStream);
- for (Entry<TTimePartitionSlot, List<TRegionReplicaSet>> timePartitionSlotListEntry :
- seriesPartitionSlotMapEntry.getValue().entrySet()) {
- timePartitionSlotListEntry.getKey().write(protocol);
- ReadWriteIOUtils.write(timePartitionSlotListEntry.getValue().size(), outputStream);
- for (TRegionReplicaSet tRegionReplicaSet : timePartitionSlotListEntry.getValue()) {
- tRegionReplicaSet.write(protocol);
- }
- }
- }
- }
- }
-
- public void deserialize(InputStream inputStream, TProtocol protocol)
- throws TException, IOException {
- int storageGroupNum = ReadWriteIOUtils.readInt(inputStream);
- // Map<StorageGroup, Map<TSeriesPartitionSlot, Map<TTimePartitionSlot, List<TRegionMessage>>>>
- while (storageGroupNum > 0) {
- String storageGroup = ReadWriteIOUtils.readString(inputStream);
- int tSeriesPartitionSlotNum = ReadWriteIOUtils.readInt(inputStream);
-
- Map<TSeriesPartitionSlot, Map<TTimePartitionSlot, List<TRegionReplicaSet>>>
- seriesPartitionSlotMapHashMap = new HashMap<>();
- while (tSeriesPartitionSlotNum > 0) {
- TSeriesPartitionSlot tSeriesPartitionSlot = new TSeriesPartitionSlot();
- tSeriesPartitionSlot.read(protocol);
-
- int tTimePartitionSlotNum = ReadWriteIOUtils.readInt(inputStream);
-
- Map<TTimePartitionSlot, List<TRegionReplicaSet>> timePartitionSlotListHashMap =
- new HashMap<>();
- while (tTimePartitionSlotNum > 0) {
- TTimePartitionSlot tTimePartitionSlot = new TTimePartitionSlot();
- tTimePartitionSlot.read(protocol);
- int size = ReadWriteIOUtils.readInt(inputStream);
-
- List<TRegionReplicaSet> tRegionMessageList = new ArrayList<>();
- while (size > 0) {
- TRegionReplicaSet tRegionReplicaSet = new TRegionReplicaSet();
- tRegionReplicaSet.read(protocol);
- tRegionMessageList.add(tRegionReplicaSet);
- size--;
- }
- timePartitionSlotListHashMap.put(tTimePartitionSlot, tRegionMessageList);
- tTimePartitionSlotNum--;
- }
-
- seriesPartitionSlotMapHashMap.put(tSeriesPartitionSlot, timePartitionSlotListHashMap);
- tSeriesPartitionSlotNum--;
- }
- dataPartitionMap.put(storageGroup, seriesPartitionSlotMapHashMap);
- storageGroupNum--;
- }
- }
-
@Override
public List<RegionReplicaSetInfo> getDistributionInfo() {
Map<TRegionReplicaSet, RegionReplicaSetInfo> distributionMap = new HashMap<>();
diff --git a/node-commons/src/main/java/org/apache/iotdb/commons/partition/DataPartitionTable.java b/node-commons/src/main/java/org/apache/iotdb/commons/partition/DataPartitionTable.java
new file mode 100644
index 0000000000..5c1efe7dce
--- /dev/null
+++ b/node-commons/src/main/java/org/apache/iotdb/commons/partition/DataPartitionTable.java
@@ -0,0 +1,183 @@
+/*
+ * 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 org.apache.iotdb.commons.utils.ThriftCommonsSerDeUtils;
+import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
+
+import org.apache.thrift.TException;
+import org.apache.thrift.protocol.TProtocol;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.nio.ByteBuffer;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+
+public class DataPartitionTable {
+
+ private final Map<TSeriesPartitionSlot, SeriesPartitionTable> dataPartitionMap;
+
+ public DataPartitionTable() {
+ this.dataPartitionMap = new ConcurrentHashMap<>();
+ }
+
+ public DataPartitionTable(Map<TSeriesPartitionSlot, SeriesPartitionTable> dataPartitionMap) {
+ this.dataPartitionMap = dataPartitionMap;
+ }
+
+ public Map<TSeriesPartitionSlot, SeriesPartitionTable> getDataPartitionMap() {
+ return dataPartitionMap;
+ }
+
+ /**
+ * Thread-safely get DataPartition within the specific StorageGroup
+ *
+ * @param partitionSlots SeriesPartitionSlots and TimePartitionSlots
+ * @param dataPartitionTable Store the matched Partitions
+ * @return True if all the PartitionSlots are matched, false otherwise
+ */
+ public boolean getDataPartition(
+ Map<TSeriesPartitionSlot, List<TTimePartitionSlot>> partitionSlots,
+ DataPartitionTable dataPartitionTable) {
+ AtomicBoolean result = new AtomicBoolean(true);
+ if (partitionSlots.isEmpty()) {
+ // Return all DataPartitions in one StorageGroup when the queried PartitionSlots are empty
+ dataPartitionTable.getDataPartitionMap().putAll(dataPartitionMap);
+ } else {
+ // Return the DataPartition for each SeriesPartitionSlot
+ partitionSlots.forEach(
+ (seriesPartitionSlot, timePartitionSlots) -> {
+ if (dataPartitionMap.containsKey(seriesPartitionSlot)) {
+ SeriesPartitionTable seriesPartitionTable = new SeriesPartitionTable();
+ if (!dataPartitionMap
+ .get(seriesPartitionSlot)
+ .getDataPartition(timePartitionSlots, seriesPartitionTable)) {
+ result.set(false);
+ }
+ dataPartitionTable
+ .getDataPartitionMap()
+ .put(seriesPartitionSlot, seriesPartitionTable);
+ } else {
+ result.set(false);
+ }
+ });
+ }
+ return result.get();
+ }
+
+ /**
+ * Create DataPartition within the specific StorageGroup
+ *
+ * @param assignedDataPartition Assigned result
+ * @return Number of DataPartitions added to each Region
+ */
+ public Map<TConsensusGroupId, AtomicInteger> createDataPartition(
+ DataPartitionTable assignedDataPartition) {
+ Map<TConsensusGroupId, AtomicInteger> deltaMap = new ConcurrentHashMap<>();
+
+ assignedDataPartition
+ .getDataPartitionMap()
+ .forEach(
+ ((seriesPartitionSlot, seriesPartitionTable) ->
+ dataPartitionMap
+ .computeIfAbsent(seriesPartitionSlot, empty -> new SeriesPartitionTable())
+ .createDataPartition(seriesPartitionTable, deltaMap)));
+
+ return deltaMap;
+ }
+
+ /**
+ * Only Leader use this interface. Filter unassigned DataPartitionSlots within the specific
+ * StorageGroup
+ *
+ * @param partitionSlots SeriesPartitionSlots and TimePartitionSlots
+ * @return Unassigned PartitionSlots
+ */
+ public Map<TSeriesPartitionSlot, List<TTimePartitionSlot>> filterUnassignedDataPartitionSlots(
+ Map<TSeriesPartitionSlot, List<TTimePartitionSlot>> partitionSlots) {
+ Map<TSeriesPartitionSlot, List<TTimePartitionSlot>> result = new ConcurrentHashMap<>();
+
+ partitionSlots.forEach(
+ (seriesPartitionSlot, timePartitionSlots) ->
+ result.put(
+ seriesPartitionSlot,
+ dataPartitionMap
+ .computeIfAbsent(seriesPartitionSlot, empty -> new SeriesPartitionTable())
+ .filterUnassignedDataPartitionSlots(timePartitionSlots)));
+
+ return result;
+ }
+
+ public void serialize(OutputStream outputStream, TProtocol protocol)
+ throws IOException, TException {
+ ReadWriteIOUtils.write(dataPartitionMap.size(), outputStream);
+ for (Map.Entry<TSeriesPartitionSlot, SeriesPartitionTable> seriesPartitionTableEntry :
+ dataPartitionMap.entrySet()) {
+ seriesPartitionTableEntry.getKey().write(protocol);
+ seriesPartitionTableEntry.getValue().serialize(outputStream, protocol);
+ }
+ }
+
+ /** Only for ConsensusRequest */
+ public void deserialize(ByteBuffer buffer) {
+ int length = buffer.getInt();
+ for (int i = 0; i < length; i++) {
+ TSeriesPartitionSlot seriesPartitionSlot =
+ ThriftCommonsSerDeUtils.deserializeTSeriesPartitionSlot(buffer);
+ SeriesPartitionTable seriesPartitionTable = new SeriesPartitionTable();
+ seriesPartitionTable.deserialize(buffer);
+ dataPartitionMap.put(seriesPartitionSlot, seriesPartitionTable);
+ }
+ }
+
+ /** Only for Snapshot */
+ public void deserialize(InputStream inputStream, TProtocol protocol)
+ throws IOException, TException {
+ int length = ReadWriteIOUtils.readInt(inputStream);
+ for (int i = 0; i < length; i++) {
+ TSeriesPartitionSlot seriesPartitionSlot = new TSeriesPartitionSlot();
+ seriesPartitionSlot.read(protocol);
+ SeriesPartitionTable seriesPartitionTable = new SeriesPartitionTable();
+ seriesPartitionTable.deserialize(inputStream, protocol);
+ dataPartitionMap.put(seriesPartitionSlot, seriesPartitionTable);
+ }
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+ DataPartitionTable that = (DataPartitionTable) o;
+ return dataPartitionMap.equals(that.dataPartitionMap);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(dataPartitionMap);
+ }
+}
diff --git a/node-commons/src/main/java/org/apache/iotdb/commons/partition/SchemaPartition.java b/node-commons/src/main/java/org/apache/iotdb/commons/partition/SchemaPartition.java
index acc4174e64..989586458f 100644
--- a/node-commons/src/main/java/org/apache/iotdb/commons/partition/SchemaPartition.java
+++ b/node-commons/src/main/java/org/apache/iotdb/commons/partition/SchemaPartition.java
@@ -20,21 +20,13 @@ package org.apache.iotdb.commons.partition;
import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet;
import org.apache.iotdb.common.rpc.thrift.TSeriesPartitionSlot;
-import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
-import org.apache.thrift.TException;
-import org.apache.thrift.protocol.TProtocol;
-
-import java.io.IOException;
-import java.io.InputStream;
-import java.io.OutputStream;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
-import java.util.Map.Entry;
-import java.util.Set;
+// TODO: Remove this class
public class SchemaPartition extends Partition {
// Map<StorageGroup, Map<TSeriesPartitionSlot, TSchemaRegionPlaceInfo>>
@@ -85,152 +77,6 @@ public class SchemaPartition extends Partition {
return null;
}
- /* Interfaces for ConfigNode */
-
- /**
- * Get SchemaPartition by partitionSlotsMap
- *
- * @param partitionSlotsMap Map<StorageGroup, List<SeriesPartitionSlot>>
- * @param preDeletedStorageGroup
- * @return Subset of current SchemaPartition, including Map<StorageGroup, Map<SeriesPartitionSlot,
- * RegionReplicaSet>>
- */
- public SchemaPartition getSchemaPartition(
- Map<String, List<TSeriesPartitionSlot>> partitionSlotsMap,
- Set<String> preDeletedStorageGroup) {
- if (partitionSlotsMap.isEmpty()) {
- // Return all SchemaPartitions when the partitionSlotsMap is empty
- final Map<String, Map<TSeriesPartitionSlot, TRegionReplicaSet>> resultAll =
- new HashMap<>(schemaPartitionMap);
- for (String preDeleted : preDeletedStorageGroup) {
- resultAll.remove(preDeleted);
- }
- return new SchemaPartition(resultAll, seriesSlotExecutorName, seriesPartitionSlotNum);
- } else {
- Map<String, Map<TSeriesPartitionSlot, TRegionReplicaSet>> result = new HashMap<>();
-
- partitionSlotsMap.forEach(
- (storageGroup, seriesPartitionSlots) -> {
- if (schemaPartitionMap.containsKey(storageGroup)
- && !preDeletedStorageGroup.contains(storageGroup)) {
- if (seriesPartitionSlots.isEmpty()) {
- // Return all SchemaPartitions in one StorageGroup when the queried
- // SeriesPartitionSlots is empty
- result.put(storageGroup, new HashMap<>(schemaPartitionMap.get(storageGroup)));
- } else {
- // Return the specific SchemaPartition
- seriesPartitionSlots.forEach(
- seriesPartitionSlot -> {
- if (schemaPartitionMap.get(storageGroup).containsKey(seriesPartitionSlot)) {
- result
- .computeIfAbsent(storageGroup, key -> new HashMap<>())
- .put(
- seriesPartitionSlot,
- schemaPartitionMap.get(storageGroup).get(seriesPartitionSlot));
- }
- });
- }
- }
- });
-
- return new SchemaPartition(result, seriesSlotExecutorName, seriesPartitionSlotNum);
- }
- }
-
- /**
- * Get SchemaPartition by storageGroup name
- *
- * @param matchedStorageGroup List<String>
- * @return Subset of current SchemaPartition which contains matchedStorageGroup
- */
- public SchemaPartition getSchemaPartition(List<String> matchedStorageGroup) {
- Map<String, Map<TSeriesPartitionSlot, TRegionReplicaSet>> result = new HashMap<>();
- matchedStorageGroup.forEach(
- (storageGroup) -> {
- if (schemaPartitionMap.containsKey(storageGroup)) {
- result.put(storageGroup, new HashMap<>(schemaPartitionMap.get(storageGroup)));
- }
- });
- return new SchemaPartition(result, seriesSlotExecutorName, seriesPartitionSlotNum);
- }
-
- /**
- * Filter out unassigned PartitionSlots
- *
- * @param partitionSlotsMap Map<StorageGroupName, List<SeriesPartitionSlot>>
- * @return Map<String, List < SeriesPartitionSlot>>, unassigned PartitionSlots
- */
- public Map<String, List<TSeriesPartitionSlot>> filterNoAssignedSchemaPartitionSlot(
- Map<String, List<TSeriesPartitionSlot>> partitionSlotsMap) {
- Map<String, List<TSeriesPartitionSlot>> result = new HashMap<>();
-
- partitionSlotsMap.forEach(
- (storageGroup, seriesPartitionSlots) -> {
- // Compare StorageGroup
- if (!schemaPartitionMap.containsKey(storageGroup)) {
- result.put(storageGroup, partitionSlotsMap.get(storageGroup));
- } else {
- seriesPartitionSlots.forEach(
- seriesPartitionSlot -> {
- // Compare SeriesPartitionSlot
- if (!schemaPartitionMap.get(storageGroup).containsKey(seriesPartitionSlot)) {
- result
- .computeIfAbsent(storageGroup, key -> new ArrayList<>())
- .add(seriesPartitionSlot);
- }
- });
- }
- });
-
- return result;
- }
-
- /** Create a SchemaPartition by ConfigNode */
- public void createSchemaPartition(
- String storageGroup,
- TSeriesPartitionSlot seriesPartitionSlot,
- TRegionReplicaSet regionReplicaSet) {
- schemaPartitionMap
- .computeIfAbsent(storageGroup, key -> new HashMap<>())
- .put(seriesPartitionSlot, regionReplicaSet);
- }
-
- public void serialize(OutputStream outputStream, TProtocol protocol)
- throws IOException, TException {
- ReadWriteIOUtils.write(schemaPartitionMap.size(), outputStream);
- for (Entry<String, Map<TSeriesPartitionSlot, TRegionReplicaSet>> entry :
- schemaPartitionMap.entrySet()) {
- ReadWriteIOUtils.write(entry.getKey(), outputStream);
- writeMap(entry.getValue(), outputStream, protocol);
- }
- }
-
- public void deserialize(InputStream inputStream, TProtocol protocol)
- throws TException, IOException {
- int size = ReadWriteIOUtils.readInt(inputStream);
- while (size > 0) {
- String key = ReadWriteIOUtils.readString(inputStream);
- Map<TSeriesPartitionSlot, TRegionReplicaSet> value = readMap(inputStream, protocol);
- schemaPartitionMap.put(key, value);
- size--;
- }
- }
-
- private Map<TSeriesPartitionSlot, TRegionReplicaSet> readMap(
- InputStream inputStream, TProtocol protocol) throws TException, IOException {
- int size = ReadWriteIOUtils.readInt(inputStream);
- Map<TSeriesPartitionSlot, TRegionReplicaSet> result = new HashMap<>();
- while (size > 0) {
- TSeriesPartitionSlot tSeriesPartitionSlot = new TSeriesPartitionSlot();
- tSeriesPartitionSlot.read(protocol);
- TRegionReplicaSet tRegionReplicaSet = new TRegionReplicaSet();
- tRegionReplicaSet.read(protocol);
- result.put(tSeriesPartitionSlot, tRegionReplicaSet);
- size--;
- }
- return result;
- }
-
@Override
public List<RegionReplicaSetInfo> getDistributionInfo() {
Map<TRegionReplicaSet, RegionReplicaSetInfo> distributionMap = new HashMap<>();
@@ -244,16 +90,4 @@ public class SchemaPartition extends Partition {
});
return new ArrayList<>(distributionMap.values());
}
-
- private void writeMap(
- Map<TSeriesPartitionSlot, TRegionReplicaSet> valueMap,
- OutputStream outputStream,
- TProtocol protocol)
- throws TException, IOException {
- ReadWriteIOUtils.write(valueMap.size(), outputStream);
- for (Entry<TSeriesPartitionSlot, TRegionReplicaSet> entry : valueMap.entrySet()) {
- entry.getKey().write(protocol);
- entry.getValue().write(protocol);
- }
- }
}
diff --git a/node-commons/src/main/java/org/apache/iotdb/commons/partition/SchemaPartitionTable.java b/node-commons/src/main/java/org/apache/iotdb/commons/partition/SchemaPartitionTable.java
new file mode 100644
index 0000000000..e13f7bfc95
--- /dev/null
+++ b/node-commons/src/main/java/org/apache/iotdb/commons/partition/SchemaPartitionTable.java
@@ -0,0 +1,177 @@
+/*
+ * 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.commons.utils.ThriftCommonsSerDeUtils;
+import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
+
+import org.apache.thrift.TException;
+import org.apache.thrift.protocol.TProtocol;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.nio.ByteBuffer;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Vector;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+
+public class SchemaPartitionTable {
+
+ private final Map<TSeriesPartitionSlot, TConsensusGroupId> schemaPartitionMap;
+
+ public SchemaPartitionTable() {
+ this.schemaPartitionMap = new ConcurrentHashMap<>();
+ }
+
+ public SchemaPartitionTable(Map<TSeriesPartitionSlot, TConsensusGroupId> schemaPartitionMap) {
+ this.schemaPartitionMap = schemaPartitionMap;
+ }
+
+ public Map<TSeriesPartitionSlot, TConsensusGroupId> getSchemaPartitionMap() {
+ return schemaPartitionMap;
+ }
+
+ /**
+ * Thread-safely get SchemaPartition within the specific StorageGroup
+ *
+ * @param partitionSlots SeriesPartitionSlots
+ * @param schemaPartitionTable Store the matched SchemaPartitions
+ * @return True if all the SeriesPartitionSlots are matched, false otherwise
+ */
+ public boolean getSchemaPartition(
+ List<TSeriesPartitionSlot> partitionSlots, SchemaPartitionTable schemaPartitionTable) {
+ AtomicBoolean result = new AtomicBoolean(true);
+
+ if (partitionSlots.isEmpty()) {
+ // Return all SchemaPartitions in one StorageGroup when the queried PartitionSlots are empty
+ schemaPartitionTable.getSchemaPartitionMap().putAll(schemaPartitionMap);
+ } else {
+ // Return the SchemaPartition for each SeriesPartitionSlot
+ partitionSlots.forEach(
+ seriesPartitionSlot -> {
+ if (schemaPartitionMap.containsKey(seriesPartitionSlot)) {
+ schemaPartitionTable
+ .getSchemaPartitionMap()
+ .put(seriesPartitionSlot, schemaPartitionMap.get(seriesPartitionSlot));
+ } else {
+ result.set(false);
+ }
+ });
+ }
+
+ return result.get();
+ }
+
+ /**
+ * Create SchemaPartition within the specific StorageGroup
+ *
+ * @param assignedSchemaPartition assigned result
+ * @return Number of SchemaPartitions added to each Region
+ */
+ public Map<TConsensusGroupId, AtomicInteger> createSchemaPartition(
+ SchemaPartitionTable assignedSchemaPartition) {
+ Map<TConsensusGroupId, AtomicInteger> deltaMap = new ConcurrentHashMap<>();
+
+ assignedSchemaPartition
+ .getSchemaPartitionMap()
+ .forEach(
+ ((seriesPartitionSlot, consensusGroupId) -> {
+ schemaPartitionMap.put(seriesPartitionSlot, consensusGroupId);
+ deltaMap
+ .computeIfAbsent(consensusGroupId, empty -> new AtomicInteger(0))
+ .getAndIncrement();
+ }));
+
+ return deltaMap;
+ }
+
+ /**
+ * Only Leader use this interface. Filter unassigned SchemaPartitionSlots within the specific
+ * StorageGroup.
+ *
+ * @param partitionSlots List<TSeriesPartitionSlot>
+ * @return Unassigned PartitionSlots
+ */
+ public List<TSeriesPartitionSlot> filterUnassignedSchemaPartitionSlots(
+ List<TSeriesPartitionSlot> partitionSlots) {
+ List<TSeriesPartitionSlot> result = new Vector<>();
+
+ partitionSlots.forEach(
+ seriesPartitionSlot -> {
+ if (!schemaPartitionMap.containsKey(seriesPartitionSlot)) {
+ result.add(seriesPartitionSlot);
+ }
+ });
+
+ return result;
+ }
+
+ public void serialize(OutputStream outputStream, TProtocol protocol)
+ throws IOException, TException {
+ ReadWriteIOUtils.write(schemaPartitionMap.size(), outputStream);
+ for (Map.Entry<TSeriesPartitionSlot, TConsensusGroupId> schemaPartitionEntry :
+ schemaPartitionMap.entrySet()) {
+ schemaPartitionEntry.getKey().write(protocol);
+ schemaPartitionEntry.getValue().write(protocol);
+ }
+ }
+
+ /** Only for ConsensusRequest */
+ public void deserialize(ByteBuffer buffer) {
+ int length = buffer.getInt();
+ for (int i = 0; i < length; i++) {
+ schemaPartitionMap.put(
+ ThriftCommonsSerDeUtils.deserializeTSeriesPartitionSlot(buffer),
+ ThriftCommonsSerDeUtils.deserializeTConsensusGroupId(buffer));
+ }
+ }
+
+ /** Only for Snapshot */
+ public void deserialize(InputStream inputStream, TProtocol protocol)
+ throws IOException, TException {
+ int length = ReadWriteIOUtils.readInt(inputStream);
+ for (int i = 0; i < length; i++) {
+ TSeriesPartitionSlot seriesPartitionSlot = new TSeriesPartitionSlot();
+ seriesPartitionSlot.read(protocol);
+ TConsensusGroupId consensusGroupId = new TConsensusGroupId();
+ consensusGroupId.read(protocol);
+ schemaPartitionMap.put(seriesPartitionSlot, consensusGroupId);
+ }
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+ SchemaPartitionTable that = (SchemaPartitionTable) o;
+ return schemaPartitionMap.equals(that.schemaPartitionMap);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(schemaPartitionMap);
+ }
+}
diff --git a/node-commons/src/main/java/org/apache/iotdb/commons/partition/SeriesPartitionTable.java b/node-commons/src/main/java/org/apache/iotdb/commons/partition/SeriesPartitionTable.java
new file mode 100644
index 0000000000..7d76ec942d
--- /dev/null
+++ b/node-commons/src/main/java/org/apache/iotdb/commons/partition/SeriesPartitionTable.java
@@ -0,0 +1,192 @@
+/*
+ * 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.TTimePartitionSlot;
+import org.apache.iotdb.commons.utils.ThriftCommonsSerDeUtils;
+import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
+
+import org.apache.thrift.TException;
+import org.apache.thrift.protocol.TProtocol;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.nio.ByteBuffer;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Vector;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+
+public class SeriesPartitionTable {
+
+ private final Map<TTimePartitionSlot, List<TConsensusGroupId>> seriesPartitionMap;
+
+ public SeriesPartitionTable() {
+ this.seriesPartitionMap = new ConcurrentHashMap<>();
+ }
+
+ public SeriesPartitionTable(Map<TTimePartitionSlot, List<TConsensusGroupId>> seriesPartitionMap) {
+ this.seriesPartitionMap = seriesPartitionMap;
+ }
+
+ public Map<TTimePartitionSlot, List<TConsensusGroupId>> getSeriesPartitionMap() {
+ return seriesPartitionMap;
+ }
+
+ /**
+ * Thread-safely get DataPartition within the specific StorageGroup
+ *
+ * @param partitionSlots TimePartitionSlots
+ * @param seriesPartitionTable Store the matched SeriesPartitions
+ * @return True if all the SeriesPartitionSlots are matched, false otherwise
+ */
+ public boolean getDataPartition(
+ List<TTimePartitionSlot> partitionSlots, SeriesPartitionTable seriesPartitionTable) {
+ AtomicBoolean result = new AtomicBoolean(true);
+
+ if (partitionSlots.isEmpty()) {
+ // Return all DataPartitions in one SeriesPartitionSlot
+ // when the queried TimePartitionSlots are empty
+ seriesPartitionTable.getSeriesPartitionMap().putAll(seriesPartitionMap);
+ } else {
+ // Return the DataPartition for each TimePartitionSlot
+ partitionSlots.forEach(
+ timePartitionSlot -> {
+ if (seriesPartitionMap.containsKey(timePartitionSlot)) {
+ seriesPartitionTable
+ .getSeriesPartitionMap()
+ .put(timePartitionSlot, seriesPartitionMap.get(timePartitionSlot));
+ } else {
+ result.set(false);
+ }
+ });
+ }
+
+ return result.get();
+ }
+
+ /**
+ * Create DataPartition within the specific SeriesPartitionSlot
+ *
+ * @param assignedSeriesPartitionTable Assigned result
+ * @param deltaMap Number of DataPartitions added to each Region
+ */
+ public void createDataPartition(
+ SeriesPartitionTable assignedSeriesPartitionTable,
+ Map<TConsensusGroupId, AtomicInteger> deltaMap) {
+ assignedSeriesPartitionTable
+ .getSeriesPartitionMap()
+ .forEach(
+ ((timePartitionSlot, consensusGroupIds) -> {
+ seriesPartitionMap.put(timePartitionSlot, new Vector<>(consensusGroupIds));
+ deltaMap
+ .computeIfAbsent(consensusGroupIds.get(0), empty -> new AtomicInteger(0))
+ .getAndIncrement();
+ }));
+ }
+
+ /**
+ * Only Leader use this interface. And this interface is synchronized. Thread-safely filter no
+ * assigned DataPartitionSlots within the specific SeriesPartitionSlot
+ *
+ * @param partitionSlots TimePartitionSlots
+ * @return Unassigned PartitionSlots
+ */
+ public synchronized List<TTimePartitionSlot> filterUnassignedDataPartitionSlots(
+ List<TTimePartitionSlot> partitionSlots) {
+ List<TTimePartitionSlot> result = new Vector<>();
+
+ partitionSlots.forEach(
+ timePartitionSlot -> {
+ if (!seriesPartitionMap.containsKey(timePartitionSlot)) {
+ result.add(timePartitionSlot);
+ }
+ });
+
+ return result;
+ }
+
+ public void serialize(OutputStream outputStream, TProtocol protocol)
+ throws IOException, TException {
+ ReadWriteIOUtils.write(seriesPartitionMap.size(), outputStream);
+ for (Map.Entry<TTimePartitionSlot, List<TConsensusGroupId>> seriesPartitionEntry :
+ seriesPartitionMap.entrySet()) {
+ seriesPartitionEntry.getKey().write(protocol);
+ ReadWriteIOUtils.write(seriesPartitionEntry.getValue().size(), outputStream);
+ for (TConsensusGroupId consensusGroupId : seriesPartitionEntry.getValue()) {
+ consensusGroupId.write(protocol);
+ }
+ }
+ }
+
+ /** Only for ConsensusRequest */
+ public void deserialize(ByteBuffer buffer) {
+ int timePartitionSlotNum = buffer.getInt();
+ for (int i = 0; i < timePartitionSlotNum; i++) {
+ TTimePartitionSlot timePartitionSlot =
+ ThriftCommonsSerDeUtils.deserializeTTimePartitionSlot(buffer);
+
+ int consensusGroupIdNum = buffer.getInt();
+ List<TConsensusGroupId> consensusGroupIds = new Vector<>();
+ for (int j = 0; j < consensusGroupIdNum; j++) {
+ consensusGroupIds.add(ThriftCommonsSerDeUtils.deserializeTConsensusGroupId(buffer));
+ }
+
+ seriesPartitionMap.put(timePartitionSlot, consensusGroupIds);
+ }
+ }
+
+ /** Only for Snapshot */
+ public void deserialize(InputStream inputStream, TProtocol protocol)
+ throws IOException, TException {
+ int timePartitionSlotNum = ReadWriteIOUtils.readInt(inputStream);
+ for (int i = 0; i < timePartitionSlotNum; i++) {
+ TTimePartitionSlot timePartitionSlot = new TTimePartitionSlot();
+ timePartitionSlot.read(protocol);
+
+ int consensusGroupIdNum = ReadWriteIOUtils.readInt(inputStream);
+ List<TConsensusGroupId> consensusGroupIds = new Vector<>();
+ for (int j = 0; j < consensusGroupIdNum; j++) {
+ TConsensusGroupId consensusGroupId = new TConsensusGroupId();
+ consensusGroupId.read(protocol);
+ consensusGroupIds.add(consensusGroupId);
+ }
+
+ seriesPartitionMap.put(timePartitionSlot, consensusGroupIds);
+ }
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+ SeriesPartitionTable that = (SeriesPartitionTable) o;
+ return seriesPartitionMap.equals(that.seriesPartitionMap);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(seriesPartitionMap);
+ }
+}
diff --git a/node-commons/src/test/java/org/apache/iotdb/commons/partition/DataPartitionTableTest.java b/node-commons/src/test/java/org/apache/iotdb/commons/partition/DataPartitionTableTest.java
new file mode 100644
index 0000000000..3125c966a6
--- /dev/null
+++ b/node-commons/src/test/java/org/apache/iotdb/commons/partition/DataPartitionTableTest.java
@@ -0,0 +1,124 @@
+/*
+ * 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.TConsensusGroupType;
+import org.apache.iotdb.common.rpc.thrift.TSeriesPartitionSlot;
+import org.apache.iotdb.common.rpc.thrift.TTimePartitionSlot;
+import org.apache.iotdb.tsfile.utils.PublicBAOS;
+
+import org.apache.thrift.TException;
+import org.apache.thrift.protocol.TBinaryProtocol;
+import org.apache.thrift.transport.TIOStreamTransport;
+import org.apache.thrift.transport.TTransport;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.io.ByteArrayInputStream;
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+public class DataPartitionTableTest {
+
+ @Test
+ public void reqSerDeTest() throws TException, IOException {
+ // Open output stream
+ PublicBAOS byteArrayOutputStream = new PublicBAOS();
+ DataOutputStream outputStream = new DataOutputStream(byteArrayOutputStream);
+
+ // Open output protocol
+ TTransport transport = new TIOStreamTransport(outputStream);
+ TBinaryProtocol protocol = new TBinaryProtocol(transport);
+
+ // Create DataPartitionTable and serialize
+ Map<TSeriesPartitionSlot, SeriesPartitionTable> dataPartitionMap = new HashMap<>();
+ for (int i = 0; i < 2; i++) {
+ TSeriesPartitionSlot seriesPartitionSlot = new TSeriesPartitionSlot(i);
+ Map<TTimePartitionSlot, List<TConsensusGroupId>> seriesPartitionMap = new HashMap<>();
+ for (int j = 0; j < 4; j++) {
+ TTimePartitionSlot timePartitionSlot = new TTimePartitionSlot(j);
+ List<TConsensusGroupId> consensusGroupIds = new ArrayList<>();
+ for (int k = 0; k < 8; k++) {
+ consensusGroupIds.add(new TConsensusGroupId(TConsensusGroupType.DataRegion, k));
+ }
+ seriesPartitionMap.put(timePartitionSlot, consensusGroupIds);
+ }
+ SeriesPartitionTable seriesPartitionTable = new SeriesPartitionTable(seriesPartitionMap);
+ dataPartitionMap.put(seriesPartitionSlot, seriesPartitionTable);
+ }
+ DataPartitionTable table0 = new DataPartitionTable(dataPartitionMap);
+ table0.serialize(outputStream, protocol);
+
+ // Deserialize by ByteBuffer
+ DataPartitionTable table1 = new DataPartitionTable();
+ table1.deserialize(
+ ByteBuffer.wrap(byteArrayOutputStream.getBuf(), 0, byteArrayOutputStream.size()));
+ Assert.assertEquals(table0, table1);
+ }
+
+ @Test
+ public void snapshotSerDeTest() throws TException, IOException {
+ // Open output stream
+ PublicBAOS byteArrayOutputStream = new PublicBAOS();
+ DataOutputStream outputStream = new DataOutputStream(byteArrayOutputStream);
+
+ // Open output protocol
+ TTransport transport = new TIOStreamTransport(outputStream);
+ TBinaryProtocol protocol = new TBinaryProtocol(transport);
+
+ // Create DataPartitionTable and serialize
+ Map<TSeriesPartitionSlot, SeriesPartitionTable> dataPartitionMap = new HashMap<>();
+ for (int i = 0; i < 2; i++) {
+ TSeriesPartitionSlot seriesPartitionSlot = new TSeriesPartitionSlot(i);
+ Map<TTimePartitionSlot, List<TConsensusGroupId>> seriesPartitionMap = new HashMap<>();
+ for (int j = 0; j < 4; j++) {
+ TTimePartitionSlot timePartitionSlot = new TTimePartitionSlot(j);
+ List<TConsensusGroupId> consensusGroupIds = new ArrayList<>();
+ for (int k = 0; k < 8; k++) {
+ consensusGroupIds.add(new TConsensusGroupId(TConsensusGroupType.DataRegion, k));
+ }
+ seriesPartitionMap.put(timePartitionSlot, consensusGroupIds);
+ }
+ SeriesPartitionTable seriesPartitionTable = new SeriesPartitionTable(seriesPartitionMap);
+ dataPartitionMap.put(seriesPartitionSlot, seriesPartitionTable);
+ }
+ DataPartitionTable table0 = new DataPartitionTable(dataPartitionMap);
+ table0.serialize(outputStream, protocol);
+
+ // Open input stream
+ DataInputStream inputStream =
+ new DataInputStream(new ByteArrayInputStream(byteArrayOutputStream.getBuf()));
+
+ // Open input protocol
+ transport = new TIOStreamTransport(inputStream);
+ protocol = new TBinaryProtocol(transport);
+
+ // Deserialize by input stream and protocol
+ DataPartitionTable table1 = new DataPartitionTable();
+ table1.deserialize(inputStream, protocol);
+ Assert.assertEquals(table0, table1);
+ }
+}
diff --git a/node-commons/src/test/java/org/apache/iotdb/commons/partition/DataPartitionTest.java b/node-commons/src/test/java/org/apache/iotdb/commons/partition/DataPartitionTest.java
deleted file mode 100644
index 9f52323bff..0000000000
--- a/node-commons/src/test/java/org/apache/iotdb/commons/partition/DataPartitionTest.java
+++ /dev/null
@@ -1,69 +0,0 @@
-/*
- * 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.TRegionReplicaSet;
-import org.apache.iotdb.common.rpc.thrift.TSeriesPartitionSlot;
-import org.apache.iotdb.common.rpc.thrift.TTimePartitionSlot;
-
-import org.apache.thrift.TException;
-import org.apache.thrift.protocol.TBinaryProtocol;
-import org.apache.thrift.protocol.TProtocol;
-import org.apache.thrift.transport.TIOStreamTransport;
-import org.junit.Assert;
-import org.junit.Test;
-
-import java.io.ByteArrayInputStream;
-import java.io.ByteArrayOutputStream;
-import java.io.IOException;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-
-public class DataPartitionTest extends SerializeTest {
-
- @Test
- public void testSerialize() throws TException, IOException {
-
- int dataPartitionFlag = 10;
-
- DataPartition dataPartition = new DataPartition(seriesPartitionExecutorClass, 1);
-
- Map<String, Map<TSeriesPartitionSlot, Map<TTimePartitionSlot, List<TRegionReplicaSet>>>>
- assignedDataPartition =
- generateCreateDataPartitionMap(
- dataPartitionFlag, generateTConsensusGroupId(dataPartitionFlag));
- dataPartition.setDataPartitionMap(assignedDataPartition);
-
- ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
- TIOStreamTransport tioStreamTransport = new TIOStreamTransport(byteArrayOutputStream);
- TProtocol protocol = new TBinaryProtocol(tioStreamTransport);
- dataPartition.serialize(byteArrayOutputStream, protocol);
-
- DataPartition newDataPartition =
- new DataPartition(new HashMap<>(), seriesPartitionExecutorClass, 1);
- ByteArrayInputStream byteArrayInputStream =
- new ByteArrayInputStream(byteArrayOutputStream.toByteArray());
- tioStreamTransport = new TIOStreamTransport(byteArrayInputStream);
- protocol = new TBinaryProtocol(tioStreamTransport);
- newDataPartition.deserialize(byteArrayInputStream, protocol);
- Assert.assertEquals(assignedDataPartition, newDataPartition.getDataPartitionMap());
- }
-}
diff --git a/node-commons/src/test/java/org/apache/iotdb/commons/partition/SchemaPartitionTableTest.java b/node-commons/src/test/java/org/apache/iotdb/commons/partition/SchemaPartitionTableTest.java
new file mode 100644
index 0000000000..e8830d0c42
--- /dev/null
+++ b/node-commons/src/test/java/org/apache/iotdb/commons/partition/SchemaPartitionTableTest.java
@@ -0,0 +1,101 @@
+/*
+ * 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.TConsensusGroupType;
+import org.apache.iotdb.common.rpc.thrift.TSeriesPartitionSlot;
+import org.apache.iotdb.tsfile.utils.PublicBAOS;
+
+import org.apache.thrift.TException;
+import org.apache.thrift.protocol.TBinaryProtocol;
+import org.apache.thrift.transport.TIOStreamTransport;
+import org.apache.thrift.transport.TTransport;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.io.ByteArrayInputStream;
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.HashMap;
+import java.util.Map;
+
+public class SchemaPartitionTableTest {
+
+ @Test
+ public void reqSerDeTest() throws TException, IOException {
+ // Open output stream
+ PublicBAOS byteArrayOutputStream = new PublicBAOS();
+ DataOutputStream outputStream = new DataOutputStream(byteArrayOutputStream);
+
+ // Open output protocol
+ TTransport transport = new TIOStreamTransport(outputStream);
+ TBinaryProtocol protocol = new TBinaryProtocol(transport);
+
+ // Create SchemaPartitionTable and serialize
+ Map<TSeriesPartitionSlot, TConsensusGroupId> schemaPartitionMap = new HashMap<>();
+ for (int i = 0; i < 10; i++) {
+ schemaPartitionMap.put(
+ new TSeriesPartitionSlot(i), new TConsensusGroupId(TConsensusGroupType.SchemaRegion, i));
+ }
+ SchemaPartitionTable table0 = new SchemaPartitionTable(schemaPartitionMap);
+ table0.serialize(outputStream, protocol);
+
+ // Deserialize by ByteBuffer
+ SchemaPartitionTable table1 = new SchemaPartitionTable();
+ table1.deserialize(
+ ByteBuffer.wrap(byteArrayOutputStream.getBuf(), 0, byteArrayOutputStream.size()));
+ Assert.assertEquals(table0, table1);
+ }
+
+ @Test
+ public void snapshotSerDeTest() throws TException, IOException {
+ // Open output stream
+ PublicBAOS byteArrayOutputStream = new PublicBAOS();
+ DataOutputStream outputStream = new DataOutputStream(byteArrayOutputStream);
+
+ // Open output protocol
+ TTransport transport = new TIOStreamTransport(outputStream);
+ TBinaryProtocol protocol = new TBinaryProtocol(transport);
+
+ // Create SchemaPartitionTable and serialize
+ Map<TSeriesPartitionSlot, TConsensusGroupId> schemaPartitionMap = new HashMap<>();
+ for (int i = 0; i < 10; i++) {
+ schemaPartitionMap.put(
+ new TSeriesPartitionSlot(i), new TConsensusGroupId(TConsensusGroupType.SchemaRegion, i));
+ }
+ SchemaPartitionTable table0 = new SchemaPartitionTable(schemaPartitionMap);
+ table0.serialize(outputStream, protocol);
+
+ // Open input stream
+ DataInputStream inputStream =
+ new DataInputStream(new ByteArrayInputStream(byteArrayOutputStream.getBuf()));
+
+ // Open input protocol
+ transport = new TIOStreamTransport(inputStream);
+ protocol = new TBinaryProtocol(transport);
+
+ // Deserialize by input stream and protocol
+ SchemaPartitionTable table1 = new SchemaPartitionTable();
+ table1.deserialize(inputStream, protocol);
+ Assert.assertEquals(table0, table1);
+ }
+}
diff --git a/node-commons/src/test/java/org/apache/iotdb/commons/partition/SchemaPartitionTest.java b/node-commons/src/test/java/org/apache/iotdb/commons/partition/SchemaPartitionTest.java
deleted file mode 100644
index af8ac595f1..0000000000
--- a/node-commons/src/test/java/org/apache/iotdb/commons/partition/SchemaPartitionTest.java
+++ /dev/null
@@ -1,65 +0,0 @@
-/*
- * 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.TRegionReplicaSet;
-import org.apache.iotdb.common.rpc.thrift.TSeriesPartitionSlot;
-
-import org.apache.thrift.TException;
-import org.apache.thrift.protocol.TBinaryProtocol;
-import org.apache.thrift.protocol.TProtocol;
-import org.apache.thrift.transport.TIOStreamTransport;
-import org.junit.Assert;
-import org.junit.Test;
-
-import java.io.ByteArrayInputStream;
-import java.io.ByteArrayOutputStream;
-import java.io.IOException;
-import java.util.HashMap;
-import java.util.Map;
-
-public class SchemaPartitionTest extends SerializeTest {
-
- @Test
- public void testSerialize() throws TException, IOException {
-
- int schemaPartitionFlag = 20;
-
- SchemaPartition schemaPartition =
- new SchemaPartition(new HashMap<>(), seriesPartitionExecutorClass, 1);
- Map<String, Map<TSeriesPartitionSlot, TRegionReplicaSet>> schemaPartitionMap =
- generateCreateSchemaPartitionMap(
- schemaPartitionFlag, generateTConsensusGroupId(schemaPartitionFlag));
- schemaPartition.setSchemaPartitionMap(schemaPartitionMap);
- ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
- TIOStreamTransport tioStreamTransport = new TIOStreamTransport(byteArrayOutputStream);
- TProtocol protocol = new TBinaryProtocol(tioStreamTransport);
- schemaPartition.serialize(byteArrayOutputStream, protocol);
-
- SchemaPartition newSchemaPartition =
- new SchemaPartition(new HashMap<>(), seriesPartitionExecutorClass, 1);
- ByteArrayInputStream byteArrayInputStream =
- new ByteArrayInputStream(byteArrayOutputStream.toByteArray());
- tioStreamTransport = new TIOStreamTransport(byteArrayInputStream);
- protocol = new TBinaryProtocol(tioStreamTransport);
- newSchemaPartition.deserialize(byteArrayInputStream, protocol);
- Assert.assertEquals(schemaPartitionMap, newSchemaPartition.getSchemaPartitionMap());
- }
-}
diff --git a/node-commons/src/test/java/org/apache/iotdb/commons/partition/SeriesPartitionTableTest.java b/node-commons/src/test/java/org/apache/iotdb/commons/partition/SeriesPartitionTableTest.java
new file mode 100644
index 0000000000..ce0563d389
--- /dev/null
+++ b/node-commons/src/test/java/org/apache/iotdb/commons/partition/SeriesPartitionTableTest.java
@@ -0,0 +1,111 @@
+/*
+ * 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.TConsensusGroupType;
+import org.apache.iotdb.common.rpc.thrift.TTimePartitionSlot;
+import org.apache.iotdb.tsfile.utils.PublicBAOS;
+
+import org.apache.thrift.TException;
+import org.apache.thrift.protocol.TBinaryProtocol;
+import org.apache.thrift.transport.TIOStreamTransport;
+import org.apache.thrift.transport.TTransport;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.io.ByteArrayInputStream;
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+public class SeriesPartitionTableTest {
+
+ @Test
+ public void reqSerDeTest() throws TException, IOException {
+ // Open output stream
+ PublicBAOS byteArrayOutputStream = new PublicBAOS();
+ DataOutputStream outputStream = new DataOutputStream(byteArrayOutputStream);
+
+ // Open output protocol
+ TTransport transport = new TIOStreamTransport(outputStream);
+ TBinaryProtocol protocol = new TBinaryProtocol(transport);
+
+ // Create SeriesPartitionTable and serialize
+ Map<TTimePartitionSlot, List<TConsensusGroupId>> seriesPartitionMap = new HashMap<>();
+ for (int i = 0; i < 5; i++) {
+ TTimePartitionSlot timePartitionSlot = new TTimePartitionSlot(i);
+ List<TConsensusGroupId> consensusGroupIds = new ArrayList<>();
+ for (int j = 0; j < 10; j++) {
+ consensusGroupIds.add(new TConsensusGroupId(TConsensusGroupType.DataRegion, j));
+ }
+ seriesPartitionMap.put(timePartitionSlot, consensusGroupIds);
+ }
+ SeriesPartitionTable table0 = new SeriesPartitionTable(seriesPartitionMap);
+ table0.serialize(outputStream, protocol);
+
+ // Deserialize by ByteBuffer
+ SeriesPartitionTable table1 = new SeriesPartitionTable();
+ table1.deserialize(
+ ByteBuffer.wrap(byteArrayOutputStream.getBuf(), 0, byteArrayOutputStream.size()));
+ Assert.assertEquals(table0, table1);
+ }
+
+ @Test
+ public void snapshotSerDeTest() throws TException, IOException {
+ // Open output stream
+ PublicBAOS byteArrayOutputStream = new PublicBAOS();
+ DataOutputStream outputStream = new DataOutputStream(byteArrayOutputStream);
+
+ // Open output protocol
+ TTransport transport = new TIOStreamTransport(outputStream);
+ TBinaryProtocol protocol = new TBinaryProtocol(transport);
+
+ // Create SeriesPartitionTable and serialize
+ Map<TTimePartitionSlot, List<TConsensusGroupId>> seriesPartitionMap = new HashMap<>();
+ for (int i = 0; i < 5; i++) {
+ TTimePartitionSlot timePartitionSlot = new TTimePartitionSlot(i);
+ List<TConsensusGroupId> consensusGroupIds = new ArrayList<>();
+ for (int j = 0; j < 10; j++) {
+ consensusGroupIds.add(new TConsensusGroupId(TConsensusGroupType.DataRegion, j));
+ }
+ seriesPartitionMap.put(timePartitionSlot, consensusGroupIds);
+ }
+ SeriesPartitionTable table0 = new SeriesPartitionTable(seriesPartitionMap);
+ table0.serialize(outputStream, protocol);
+
+ // Open input stream
+ DataInputStream inputStream =
+ new DataInputStream(new ByteArrayInputStream(byteArrayOutputStream.getBuf()));
+
+ // Open input protocol
+ transport = new TIOStreamTransport(inputStream);
+ protocol = new TBinaryProtocol(transport);
+
+ // Deserialize by input stream and protocol
+ SeriesPartitionTable table1 = new SeriesPartitionTable();
+ table1.deserialize(inputStream, protocol);
+ Assert.assertEquals(table0, table1);
+ }
+}
diff --git a/node-commons/src/test/java/org/apache/iotdb/commons/utils/ThriftConfigNodeSerDeUtilsTest.java b/node-commons/src/test/java/org/apache/iotdb/commons/utils/ThriftConfigNodeSerDeUtilsTest.java
index 60e0adef7b..d14de95682 100644
--- a/node-commons/src/test/java/org/apache/iotdb/commons/utils/ThriftConfigNodeSerDeUtilsTest.java
+++ b/node-commons/src/test/java/org/apache/iotdb/commons/utils/ThriftConfigNodeSerDeUtilsTest.java
@@ -19,8 +19,6 @@
package org.apache.iotdb.commons.utils;
import org.apache.iotdb.common.rpc.thrift.TConfigNodeLocation;
-import org.apache.iotdb.common.rpc.thrift.TConsensusGroupId;
-import org.apache.iotdb.common.rpc.thrift.TConsensusGroupType;
import org.apache.iotdb.common.rpc.thrift.TEndPoint;
import org.apache.iotdb.confignode.rpc.thrift.TStorageGroupSchema;
@@ -29,7 +27,6 @@ import org.junit.Assert;
import org.junit.Test;
import java.nio.ByteBuffer;
-import java.util.ArrayList;
public class ThriftConfigNodeSerDeUtilsTest {
@@ -49,17 +46,6 @@ public class ThriftConfigNodeSerDeUtilsTest {
storageGroupSchema0.setDataReplicationFactor(3);
storageGroupSchema0.setTimePartitionInterval(604800);
- storageGroupSchema0.setSchemaRegionGroupIds(new ArrayList<>());
- storageGroupSchema0.setDataRegionGroupIds(new ArrayList<>());
- for (int i = 0; i < 3; i++) {
- storageGroupSchema0
- .getSchemaRegionGroupIds()
- .add(new TConsensusGroupId(TConsensusGroupType.SchemaRegion, i * 2));
- storageGroupSchema0
- .getDataRegionGroupIds()
- .add(new TConsensusGroupId(TConsensusGroupType.DataRegion, i * 2 + 1));
- }
-
ThriftConfigNodeSerDeUtils.serializeTStorageGroupSchema(storageGroupSchema0, buffer);
buffer.flip();
TStorageGroupSchema storageGroupSchema1 =
diff --git a/service-rpc/src/main/java/org/apache/iotdb/rpc/TSStatusCode.java b/service-rpc/src/main/java/org/apache/iotdb/rpc/TSStatusCode.java
index 85ce2f54b9..2c04189ef3 100644
--- a/service-rpc/src/main/java/org/apache/iotdb/rpc/TSStatusCode.java
+++ b/service-rpc/src/main/java/org/apache/iotdb/rpc/TSStatusCode.java
@@ -124,11 +124,11 @@ public enum TSStatusCode {
// ConfigNode response
DATANODE_ALREADY_REGISTERED(901),
- STORAGE_GROUP_ALREADY_EXISTS(902),
- NOT_ENOUGH_DATA_NODE(903),
- ERROR_GLOBAL_CONFIG(904),
- APPLY_CONFIGNODE_FAILED(905),
- PERSISTENCE_FAILURE(906);
+ SET_STORAGE_GROUP_FAILED(902),
+ STORAGE_GROUP_ALREADY_EXISTS(903),
+ NOT_ENOUGH_DATA_NODE(904),
+ ERROR_GLOBAL_CONFIG(905),
+ APPLY_CONFIGNODE_FAILED(906);
private int statusCode;
diff --git a/thrift-confignode/src/main/thrift/confignode.thrift b/thrift-confignode/src/main/thrift/confignode.thrift
index 52e8d3f004..8f55224075 100644
--- a/thrift-confignode/src/main/thrift/confignode.thrift
+++ b/thrift-confignode/src/main/thrift/confignode.thrift
@@ -103,8 +103,6 @@ struct TStorageGroupSchema {
5: optional i64 timePartitionInterval
6: optional i32 maximumSchemaRegionCount
7: optional i32 maximumDataRegionCount
- 8: optional list<common.TConsensusGroupId> dataRegionGroupIds
- 9: optional list<common.TConsensusGroupId> schemaRegionGroupIds
}
// Schema
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/utils/Pair.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/utils/Pair.java
index 140d43bc0a..b51a3b4078 100644
--- a/tsfile/src/main/java/org/apache/iotdb/tsfile/utils/Pair.java
+++ b/tsfile/src/main/java/org/apache/iotdb/tsfile/utils/Pair.java
@@ -38,10 +38,18 @@ public class Pair<L, R> implements Serializable {
right = r;
}
+ public void setLeft(L left) {
+ this.left = left;
+ }
+
public L getLeft() {
return left;
}
+ public void setRight(R right) {
+ this.right = right;
+ }
+
public R getRight() {
return right;
}