This is an automated email from the ASF dual-hosted git repository.
tanxinyu pushed a commit to branch new_cluster
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/new_cluster by this push:
new 7cff204 [To new_cluster] [IOTDB-2683] Hash algorithm for device group
(#5188)
7cff204 is described below
commit 7cff204c008d6dffb70766b5324ce7ed65141238
Author: CRZbulabula <[email protected]>
AuthorDate: Fri Mar 11 09:22:59 2022 +0800
[To new_cluster] [IOTDB-2683] Hash algorithm for device group (#5188)
* stash
update proporties file tomorrow
* ConfigNode conf file
* Add annotation
* Add license
* remove useless mvn dependence
* Use refraction instead of switch
* Add device_group_hash_executor_package parameter
* merge parameters
Co-authored-by: CRZbulabula <[email protected]>
---
confignode/pom.xml | 57 +++++++++-
confignode/src/assembly/confignode.xml | 40 +++++++
.../resources/conf/iotdb-confignode.properties | 37 +++++++
.../iotdb/confignode/conf/ConfigNodeConf.java | 51 +++++++++
.../iotdb/confignode/conf/ConfigNodeConstant.java | 24 ++---
.../confignode/conf/ConfigNodeDescriptor.java | 117 +++++++++++++++++++++
.../iotdb/confignode/manager/ConfigManager.java | 43 +++++++-
.../confignode/manager/hash/APHashExecutor.java | 44 ++++----
.../confignode/manager/hash/BKDRHashExecutor.java | 30 +++---
.../manager/hash/DeviceGroupHashExecutor.java | 25 ++---
.../iotdb/confignode/manager/hash/JSHash.java | 30 +++---
.../iotdb/confignode/manager/hash/SDBMHash.java | 30 +++---
.../iotdb/confignode/partition/PartitionTable.java | 16 +--
.../service/basic/ConfigServiceProvider.java | 25 +----
.../service/thrift/impl/ConfigServiceImpl.java | 23 ++--
.../confignode/conf/ConfigNodeDescriptorTest.java | 76 +++++++++++++
.../manager/hash/DeviceGroupHashExecutorTest.java | 95 +++++++++++++++++
.../src/test/resources/iotdb-confignode.properties | 31 ++++++
pom.xml | 2 +-
.../src/main/thrift/confignode.thrift | 12 +--
20 files changed, 653 insertions(+), 155 deletions(-)
diff --git a/confignode/pom.xml b/confignode/pom.xml
index bfc5af2..feab079 100644
--- a/confignode/pom.xml
+++ b/confignode/pom.xml
@@ -29,6 +29,11 @@
</parent>
<artifactId>iotdb-confignode</artifactId>
<name>IoTDB ConfigNode</name>
+ <properties>
+ <iotdb.test.skip>false</iotdb.test.skip>
+ <iotdb.it.skip>${iotdb.test.skip}</iotdb.it.skip>
+ <iotdb.ut.skip>${iotdb.test.skip}</iotdb.ut.skip>
+ </properties>
<dependencies>
<dependency>
<groupId>org.apache.iotdb</groupId>
@@ -37,13 +42,57 @@
</dependency>
<dependency>
<groupId>org.apache.iotdb</groupId>
- <artifactId>iotdb-server</artifactId>
+ <artifactId>iotdb-thrift-confignode</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
- <groupId>org.apache.iotdb</groupId>
- <artifactId>iotdb-thrift-confignode</artifactId>
- <version>${project.version}</version>
+ <groupId>org.apache.tomcat.embed</groupId>
+ <artifactId>tomcat-embed-core</artifactId>
+ <version>8.5.46</version>
+ <scope>test</scope>
</dependency>
</dependencies>
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-assembly-plugin</artifactId>
+ <version>${maven.assembly.version}</version>
+ <executions>
+ <!-- Package binaries-->
+ <execution>
+ <id>confignode-assembly</id>
+ <phase>package</phase>
+ <goals>
+ <goal>single</goal>
+ </goals>
+ <configuration>
+ <descriptors>
+
<descriptor>src/assembly/confignode.xml</descriptor>
+ </descriptors>
+ <appendAssemblyId>false</appendAssemblyId>
+ <archive>
+ <manifest>
+
<addDefaultImplementationEntries>true</addDefaultImplementationEntries>
+
<addDefaultSpecificationEntries>true</addDefaultSpecificationEntries>
+ </manifest>
+ </archive>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+ <!-- generate test codes into test-jar-->
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-jar-plugin</artifactId>
+ <executions>
+ <execution>
+ <goals>
+ <goal>test-jar</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
</project>
diff --git a/confignode/src/assembly/confignode.xml
b/confignode/src/assembly/confignode.xml
new file mode 100644
index 0000000..5570748
--- /dev/null
+++ b/confignode/src/assembly/confignode.xml
@@ -0,0 +1,40 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+ 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.
+
+-->
+<assembly>
+ <id>confignode</id>
+ <formats>
+ <format>dir</format>
+ <format>zip</format>
+ </formats>
+ <includeBaseDirectory>false</includeBaseDirectory>
+ <dependencySets>
+ <dependencySet>
+ <outputDirectory>lib</outputDirectory>
+ </dependencySet>
+ </dependencySets>
+ <fileSets>
+ <fileSet>
+ <directory>src/assembly/resources</directory>
+ <outputDirectory>${file.separator}</outputDirectory>
+ </fileSet>
+ </fileSets>
+</assembly>
diff --git a/confignode/src/assembly/resources/conf/iotdb-confignode.properties
b/confignode/src/assembly/resources/conf/iotdb-confignode.properties
new file mode 100644
index 0000000..0dcb8b9
--- /dev/null
+++ b/confignode/src/assembly/resources/conf/iotdb-confignode.properties
@@ -0,0 +1,37 @@
+#
+# 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.
+#
+
+####################
+### DeviceGroup Configuration
+####################
+
+# Number of DeviceGroups per StorageGroup
+# Datatype: int
+# device_group_count=10000
+
+# DeviceGroup hash algorithm
+# Datatype: String
+# These hashing algorithms are currently supported:
+# 1. org.apache.iotdb.confignode.manager.hash.BKDRHashExecutor(Default)
+# 2. org.apache.iotdb.confignode.manager.hash.APHashExecutor
+# 3. org.apache.iotdb.confignode.manager.hash.JSHashExecutor
+# 4. org.apache.iotdb.confignode.manager.hash.SDBMHashExecutor
+# Also, if you want to implement your own hash algorithm, you can inherit the
DeviceGroupHashExecutor class and
+# modify this parameter to correspond to your Java class
+#
device_group_hash_executor_class=org.apache.iotdb.confignode.manager.hash.BKDRHashExecutor
\ No newline at end of file
diff --git
a/confignode/src/main/java/org/apache/iotdb/confignode/conf/ConfigNodeConf.java
b/confignode/src/main/java/org/apache/iotdb/confignode/conf/ConfigNodeConf.java
new file mode 100644
index 0000000..746512b
--- /dev/null
+++
b/confignode/src/main/java/org/apache/iotdb/confignode/conf/ConfigNodeConf.java
@@ -0,0 +1,51 @@
+/*
+ * 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.conf;
+
+public class ConfigNodeConf {
+
+ public static final String CONF_NAME = "iotdb-confignode.properties";
+
+ // Number of DeviceGroups per StorageGroup
+ private int deviceGroupCount = 10000;
+
+ // DeviceGroup hash executor class
+ private String deviceGroupHashExecutorClass =
+ "org.apache.iotdb.confignode.manager.hash.BKDRHashExecutor";
+
+ public ConfigNodeConf() {
+ // empty constructor
+ }
+
+ public int getDeviceGroupCount() {
+ return deviceGroupCount;
+ }
+
+ public void setDeviceGroupCount(int deviceGroupCount) {
+ this.deviceGroupCount = deviceGroupCount;
+ }
+
+ public String getDeviceGroupHashExecutorClass() {
+ return deviceGroupHashExecutorClass;
+ }
+
+ public void setDeviceGroupHashExecutorClass(String
deviceGroupHashExecutorClass) {
+ this.deviceGroupHashExecutorClass = deviceGroupHashExecutorClass;
+ }
+}
diff --git a/thrift-confignode/src/main/thrift/confignode.thrift
b/confignode/src/main/java/org/apache/iotdb/confignode/conf/ConfigNodeConstant.java
similarity index 55%
copy from thrift-confignode/src/main/thrift/confignode.thrift
copy to
confignode/src/main/java/org/apache/iotdb/confignode/conf/ConfigNodeConstant.java
index ba71a0e..0680a8e 100644
--- a/thrift-confignode/src/main/thrift/confignode.thrift
+++
b/confignode/src/main/java/org/apache/iotdb/confignode/conf/ConfigNodeConstant.java
@@ -16,23 +16,15 @@
* specific language governing permissions and limitations
* under the License.
*/
+package org.apache.iotdb.confignode.conf;
-include "rpc.thrift"
-namespace java org.apache.iotdb.confignode.rpc.thrift
-namespace py iotdb.thrift.confignode
+public class ConfigNodeConstant {
-service ConfigIService {
- rpc.TSStatus setStorageGroup(1:i64 sessionId, 2:string storageGroup)
+ private ConfigNodeConstant() {
+ // empty constructor
+ }
- rpc.TSStatus deleteStorageGroup(1:i64 sessionId, 2:list<string>
storageGroups)
+ public static final String CONFIG_NODE_CONF = "CONFIG_NODE_CONF";
- rpc.TSStatus createTimeseries(1:rpc.TSCreateTimeseriesReq req)
-
- rpc.TSStatus createAlignedTimeseries(1:rpc.TSCreateAlignedTimeseriesReq req)
-
- rpc.TSStatus createMultiTimeseries(1:rpc.TSCreateMultiTimeseriesReq req)
-
- rpc.TSStatus deleteTimeSeries(1:i64 sessionId, 2:list<string> paths)
-
- i32 getDeviceGroupID(1:i64 sessionId, 2:string device)
-}
\ No newline at end of file
+ public static final String CONFIG_NODE_HOME = "CONFIG_NODE_HOME";
+}
diff --git
a/confignode/src/main/java/org/apache/iotdb/confignode/conf/ConfigNodeDescriptor.java
b/confignode/src/main/java/org/apache/iotdb/confignode/conf/ConfigNodeDescriptor.java
new file mode 100644
index 0000000..0477f6d
--- /dev/null
+++
b/confignode/src/main/java/org/apache/iotdb/confignode/conf/ConfigNodeDescriptor.java
@@ -0,0 +1,117 @@
+/*
+ * 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.conf;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+import java.net.MalformedURLException;
+import java.net.URL;
+import java.util.Properties;
+
+public class ConfigNodeDescriptor {
+
+ private static final Logger LOGGER =
LoggerFactory.getLogger(ConfigNodeDescriptor.class);
+
+ private final ConfigNodeConf conf = new ConfigNodeConf();
+
+ private ConfigNodeDescriptor() {
+ loadProps();
+ }
+
+ public ConfigNodeConf getConf() {
+ return conf;
+ }
+
+ public static ConfigNodeDescriptor getInstance() {
+ return ConfigNodeDescriptorHolder.INSTANCE;
+ }
+
+ private static class ConfigNodeDescriptorHolder {
+
+ private static final ConfigNodeDescriptor INSTANCE = new
ConfigNodeDescriptor();
+
+ private ConfigNodeDescriptorHolder() {
+ // empty constructor
+ }
+ }
+
+ public URL getPropsUrl() {
+ // The same logic as IoTDBDescriptor
+ String url = System.getProperty(ConfigNodeConstant.CONFIG_NODE_CONF, null);
+ if (url == null) {
+ url = System.getProperty(ConfigNodeConstant.CONFIG_NODE_HOME, null);
+ if (url != null) {
+ url = url + File.separatorChar + "conf" + File.separatorChar +
ConfigNodeConf.CONF_NAME;
+ } else {
+ URL uri = ConfigNodeConf.class.getResource("/" +
ConfigNodeConf.CONF_NAME);
+ if (uri != null) {
+ return uri;
+ }
+ LOGGER.warn(
+ "Cannot find IOTDB_HOME or IOTDB_CONF environment variable when
loading config file {}, use default configuration",
+ ConfigNodeConf.CONF_NAME);
+ return null;
+ }
+ } else if (!url.endsWith(".properties")) {
+ url += File.separator + ConfigNodeConf.CONF_NAME;
+ }
+
+ if (!url.startsWith("file:") && !url.startsWith("classpath:")) {
+ url = "file:" + url;
+ }
+
+ try {
+ return new URL(url);
+ } catch (MalformedURLException e) {
+ return null;
+ }
+ }
+
+ private void loadProps() {
+ URL url = getPropsUrl();
+ if (url == null) {
+ LOGGER.warn("Couldn't load the ConfigNode configuration from any of the
known sources.");
+ return;
+ }
+
+ try (InputStream inputStream = url.openStream()) {
+
+ LOGGER.info("start reading ConfigNode conf file: {}", url);
+
+ Properties properties = new Properties();
+ properties.load(inputStream);
+
+ conf.setDeviceGroupCount(
+ Integer.parseInt(
+ properties.getProperty(
+ "device_group_count",
String.valueOf(conf.getDeviceGroupCount()))));
+
+ conf.setDeviceGroupHashExecutorClass(
+ properties.getProperty(
+ "device_group_hash_executor_class",
conf.getDeviceGroupHashExecutorClass()));
+
+ } catch (IOException e) {
+ LOGGER.warn("Couldn't load ConfigNode conf file, use default config", e);
+ }
+ }
+}
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 21d29b9..d000c19 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
@@ -18,9 +18,17 @@
*/
package org.apache.iotdb.confignode.manager;
+import org.apache.iotdb.confignode.conf.ConfigNodeConf;
+import org.apache.iotdb.confignode.conf.ConfigNodeDescriptor;
+import org.apache.iotdb.confignode.manager.hash.DeviceGroupHashExecutor;
import org.apache.iotdb.confignode.partition.PartitionTable;
import org.apache.iotdb.confignode.service.balancer.LoadBalancer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.lang.reflect.Constructor;
+import java.lang.reflect.InvocationTargetException;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
@@ -30,20 +38,47 @@ import java.util.concurrent.locks.ReentrantLock;
*/
public class ConfigManager {
- private final Lock partitionTableLock;
- private final PartitionTable partitionTable;
+ private static final Logger LOGGER =
LoggerFactory.getLogger(ConfigManager.class);
+
+ private DeviceGroupHashExecutor hashExecutor;
+
+ private Lock partitionTableLock;
+ private PartitionTable partitionTable;
- private final LoadBalancer loadBalancer;
+ private LoadBalancer loadBalancer;
+
+ public ConfigManager(String hashExecutorClass, int deviceGroupCount) {
+ setHashExecutor(hashExecutorClass, deviceGroupCount);
+ }
public ConfigManager() {
+ ConfigNodeConf config = ConfigNodeDescriptor.getInstance().getConf();
+
+ setHashExecutor(config.getDeviceGroupHashExecutorClass(),
config.getDeviceGroupCount());
+
this.partitionTableLock = new ReentrantLock();
this.partitionTable = new PartitionTable();
this.loadBalancer = new LoadBalancer(partitionTableLock, partitionTable);
}
+ private void setHashExecutor(String hashExecutorClass, int deviceGroupCount)
{
+ try {
+ Class<?> executor = Class.forName(hashExecutorClass);
+ Constructor<?> executorConstructor = executor.getConstructor(int.class);
+ hashExecutor = (DeviceGroupHashExecutor)
executorConstructor.newInstance(deviceGroupCount);
+ } catch (ClassNotFoundException
+ | NoSuchMethodException
+ | InstantiationException
+ | IllegalAccessException
+ | InvocationTargetException e) {
+ LOGGER.error("Couldn't Constructor DeviceGroupHashExecutor class: {}",
hashExecutorClass, e);
+ hashExecutor = null;
+ }
+ }
+
public int getDeviceGroupID(String device) {
- return -1;
+ return hashExecutor.getDeviceGroupID(device);
}
// TODO: Interfaces for metadata operations
diff --git a/thrift-confignode/src/main/thrift/confignode.thrift
b/confignode/src/main/java/org/apache/iotdb/confignode/manager/hash/APHashExecutor.java
similarity index 56%
copy from thrift-confignode/src/main/thrift/confignode.thrift
copy to
confignode/src/main/java/org/apache/iotdb/confignode/manager/hash/APHashExecutor.java
index ba71a0e..0989fbb 100644
--- a/thrift-confignode/src/main/thrift/confignode.thrift
+++
b/confignode/src/main/java/org/apache/iotdb/confignode/manager/hash/APHashExecutor.java
@@ -16,23 +16,27 @@
* specific language governing permissions and limitations
* under the License.
*/
-
-include "rpc.thrift"
-namespace java org.apache.iotdb.confignode.rpc.thrift
-namespace py iotdb.thrift.confignode
-
-service ConfigIService {
- rpc.TSStatus setStorageGroup(1:i64 sessionId, 2:string storageGroup)
-
- rpc.TSStatus deleteStorageGroup(1:i64 sessionId, 2:list<string>
storageGroups)
-
- rpc.TSStatus createTimeseries(1:rpc.TSCreateTimeseriesReq req)
-
- rpc.TSStatus createAlignedTimeseries(1:rpc.TSCreateAlignedTimeseriesReq req)
-
- rpc.TSStatus createMultiTimeseries(1:rpc.TSCreateMultiTimeseriesReq req)
-
- rpc.TSStatus deleteTimeSeries(1:i64 sessionId, 2:list<string> paths)
-
- i32 getDeviceGroupID(1:i64 sessionId, 2:string device)
-}
\ No newline at end of file
+package org.apache.iotdb.confignode.manager.hash;
+
+public class APHashExecutor extends DeviceGroupHashExecutor {
+
+ public APHashExecutor(int deviceGroupCount) {
+ super(deviceGroupCount);
+ }
+
+ @Override
+ public int getDeviceGroupID(String device) {
+ int hash = 0;
+
+ for (int i = 0; i < device.length(); i++) {
+ if ((i & 1) == 0) {
+ hash ^= ((hash << 7) ^ (int) device.charAt(i) ^ (hash >> 3));
+ } else {
+ hash ^= (~((hash << 11) ^ (int) device.charAt(i) ^ (hash >> 5)));
+ }
+ }
+ hash &= Integer.MAX_VALUE;
+
+ return hash % deviceGroupCount;
+ }
+}
diff --git a/thrift-confignode/src/main/thrift/confignode.thrift
b/confignode/src/main/java/org/apache/iotdb/confignode/manager/hash/BKDRHashExecutor.java
similarity index 56%
copy from thrift-confignode/src/main/thrift/confignode.thrift
copy to
confignode/src/main/java/org/apache/iotdb/confignode/manager/hash/BKDRHashExecutor.java
index ba71a0e..2a3eed2 100644
--- a/thrift-confignode/src/main/thrift/confignode.thrift
+++
b/confignode/src/main/java/org/apache/iotdb/confignode/manager/hash/BKDRHashExecutor.java
@@ -16,23 +16,25 @@
* specific language governing permissions and limitations
* under the License.
*/
+package org.apache.iotdb.confignode.manager.hash;
-include "rpc.thrift"
-namespace java org.apache.iotdb.confignode.rpc.thrift
-namespace py iotdb.thrift.confignode
+public class BKDRHashExecutor extends DeviceGroupHashExecutor {
-service ConfigIService {
- rpc.TSStatus setStorageGroup(1:i64 sessionId, 2:string storageGroup)
+ private static final int seed = 131;
- rpc.TSStatus deleteStorageGroup(1:i64 sessionId, 2:list<string>
storageGroups)
+ public BKDRHashExecutor(int deviceGroupCount) {
+ super(deviceGroupCount);
+ }
- rpc.TSStatus createTimeseries(1:rpc.TSCreateTimeseriesReq req)
+ @Override
+ public int getDeviceGroupID(String device) {
+ int hash = 0;
- rpc.TSStatus createAlignedTimeseries(1:rpc.TSCreateAlignedTimeseriesReq req)
+ for (int i = 0; i < device.length(); i++) {
+ hash = hash * seed + (int) device.charAt(i);
+ }
+ hash &= Integer.MAX_VALUE;
- rpc.TSStatus createMultiTimeseries(1:rpc.TSCreateMultiTimeseriesReq req)
-
- rpc.TSStatus deleteTimeSeries(1:i64 sessionId, 2:list<string> paths)
-
- i32 getDeviceGroupID(1:i64 sessionId, 2:string device)
-}
\ No newline at end of file
+ return hash % deviceGroupCount;
+ }
+}
diff --git a/thrift-confignode/src/main/thrift/confignode.thrift
b/confignode/src/main/java/org/apache/iotdb/confignode/manager/hash/DeviceGroupHashExecutor.java
similarity index 55%
copy from thrift-confignode/src/main/thrift/confignode.thrift
copy to
confignode/src/main/java/org/apache/iotdb/confignode/manager/hash/DeviceGroupHashExecutor.java
index ba71a0e..bd62834 100644
--- a/thrift-confignode/src/main/thrift/confignode.thrift
+++
b/confignode/src/main/java/org/apache/iotdb/confignode/manager/hash/DeviceGroupHashExecutor.java
@@ -16,23 +16,16 @@
* specific language governing permissions and limitations
* under the License.
*/
+package org.apache.iotdb.confignode.manager.hash;
-include "rpc.thrift"
-namespace java org.apache.iotdb.confignode.rpc.thrift
-namespace py iotdb.thrift.confignode
+/** All DeviceGroup hash algorithm executors must be subclasses of
DeviceGroupHashExecutor */
+public abstract class DeviceGroupHashExecutor {
-service ConfigIService {
- rpc.TSStatus setStorageGroup(1:i64 sessionId, 2:string storageGroup)
+ protected final int deviceGroupCount;
- rpc.TSStatus deleteStorageGroup(1:i64 sessionId, 2:list<string>
storageGroups)
+ public DeviceGroupHashExecutor(int deviceGroupCount) {
+ this.deviceGroupCount = deviceGroupCount;
+ }
- rpc.TSStatus createTimeseries(1:rpc.TSCreateTimeseriesReq req)
-
- rpc.TSStatus createAlignedTimeseries(1:rpc.TSCreateAlignedTimeseriesReq req)
-
- rpc.TSStatus createMultiTimeseries(1:rpc.TSCreateMultiTimeseriesReq req)
-
- rpc.TSStatus deleteTimeSeries(1:i64 sessionId, 2:list<string> paths)
-
- i32 getDeviceGroupID(1:i64 sessionId, 2:string device)
-}
\ No newline at end of file
+ public abstract int getDeviceGroupID(String device);
+}
diff --git a/thrift-confignode/src/main/thrift/confignode.thrift
b/confignode/src/main/java/org/apache/iotdb/confignode/manager/hash/JSHash.java
similarity index 56%
copy from thrift-confignode/src/main/thrift/confignode.thrift
copy to
confignode/src/main/java/org/apache/iotdb/confignode/manager/hash/JSHash.java
index ba71a0e..8dc5664 100644
--- a/thrift-confignode/src/main/thrift/confignode.thrift
+++
b/confignode/src/main/java/org/apache/iotdb/confignode/manager/hash/JSHash.java
@@ -16,23 +16,25 @@
* specific language governing permissions and limitations
* under the License.
*/
+package org.apache.iotdb.confignode.manager.hash;
-include "rpc.thrift"
-namespace java org.apache.iotdb.confignode.rpc.thrift
-namespace py iotdb.thrift.confignode
+public class JSHash extends DeviceGroupHashExecutor {
-service ConfigIService {
- rpc.TSStatus setStorageGroup(1:i64 sessionId, 2:string storageGroup)
+ private static final int base = 1315423911;
- rpc.TSStatus deleteStorageGroup(1:i64 sessionId, 2:list<string>
storageGroups)
+ public JSHash(int deviceGroupCount) {
+ super(deviceGroupCount);
+ }
- rpc.TSStatus createTimeseries(1:rpc.TSCreateTimeseriesReq req)
+ @Override
+ public int getDeviceGroupID(String device) {
+ int hash = base;
- rpc.TSStatus createAlignedTimeseries(1:rpc.TSCreateAlignedTimeseriesReq req)
+ for (int i = 0; i < device.length(); i++) {
+ hash ^= ((hash << 5) + (int) device.charAt(i) + (hash >> 2));
+ }
+ hash &= Integer.MAX_VALUE;
- rpc.TSStatus createMultiTimeseries(1:rpc.TSCreateMultiTimeseriesReq req)
-
- rpc.TSStatus deleteTimeSeries(1:i64 sessionId, 2:list<string> paths)
-
- i32 getDeviceGroupID(1:i64 sessionId, 2:string device)
-}
\ No newline at end of file
+ return hash % deviceGroupCount;
+ }
+}
diff --git a/thrift-confignode/src/main/thrift/confignode.thrift
b/confignode/src/main/java/org/apache/iotdb/confignode/manager/hash/SDBMHash.java
similarity index 56%
copy from thrift-confignode/src/main/thrift/confignode.thrift
copy to
confignode/src/main/java/org/apache/iotdb/confignode/manager/hash/SDBMHash.java
index ba71a0e..ab528af 100644
--- a/thrift-confignode/src/main/thrift/confignode.thrift
+++
b/confignode/src/main/java/org/apache/iotdb/confignode/manager/hash/SDBMHash.java
@@ -16,23 +16,23 @@
* specific language governing permissions and limitations
* under the License.
*/
+package org.apache.iotdb.confignode.manager.hash;
-include "rpc.thrift"
-namespace java org.apache.iotdb.confignode.rpc.thrift
-namespace py iotdb.thrift.confignode
+public class SDBMHash extends DeviceGroupHashExecutor {
-service ConfigIService {
- rpc.TSStatus setStorageGroup(1:i64 sessionId, 2:string storageGroup)
+ public SDBMHash(int deviceGroupCount) {
+ super(deviceGroupCount);
+ }
- rpc.TSStatus deleteStorageGroup(1:i64 sessionId, 2:list<string>
storageGroups)
+ @Override
+ public int getDeviceGroupID(String device) {
+ int hash = 0;
- rpc.TSStatus createTimeseries(1:rpc.TSCreateTimeseriesReq req)
+ for (int i = 0; i < device.length(); i++) {
+ hash = ((int) device.charAt(i) + (hash << 6) + (hash << 16) - hash);
+ }
+ hash &= Integer.MAX_VALUE;
- rpc.TSStatus createAlignedTimeseries(1:rpc.TSCreateAlignedTimeseriesReq req)
-
- rpc.TSStatus createMultiTimeseries(1:rpc.TSCreateMultiTimeseriesReq req)
-
- rpc.TSStatus deleteTimeSeries(1:i64 sessionId, 2:list<string> paths)
-
- i32 getDeviceGroupID(1:i64 sessionId, 2:string device)
-}
\ No newline at end of file
+ return hash % deviceGroupCount;
+ }
+}
diff --git
a/confignode/src/main/java/org/apache/iotdb/confignode/partition/PartitionTable.java
b/confignode/src/main/java/org/apache/iotdb/confignode/partition/PartitionTable.java
index f6a6ec3..63cc10d 100644
---
a/confignode/src/main/java/org/apache/iotdb/confignode/partition/PartitionTable.java
+++
b/confignode/src/main/java/org/apache/iotdb/confignode/partition/PartitionTable.java
@@ -28,25 +28,25 @@ import java.util.Map;
*/
public class PartitionTable {
- // Map<StorageGroup, Map<DeviceGroupID, MManagerGroupID>>
+ // Map<StorageGroup, Map<DeviceGroupID, SchemaRegionGroupID>>
private final Map<String, Map<Integer, Integer>> metadataPartitionTable;
- // Map<MManagerGroupID, List<PhysicalNodeIDs>>
- private final Map<Integer, List<Integer>> mManagerGroupPhysicalNodesMap;
+ // Map<SchemaRegionGroupID, List<DataNodeID>>
+ private final Map<Integer, List<Integer>> schemaRegionGroupDataNodesMap;
- // Map<StorageGroup, Map<DeviceGroupID, Map<TimeInterval, List<VSGGroupID>>>>
+ // Map<StorageGroup, Map<DeviceGroupID, Map<TimeInterval,
List<DataRegionGroupID>>>>
private final Map<String, Map<Integer, Map<Long, List<Integer>>>>
dataPartitionTable;
- // Map<VSGGroupID, List<PhysicalNodeIDs>>
- private final Map<Integer, List<Integer>> vsgGroupPhysicalNodesMap;
+ // Map<DataRegionGroupID, List<DataNodeID>>
+ private final Map<Integer, List<Integer>> dataRegionGroupDataNodesMap;
// Map<StorageGroup, Map<DeviceGroupID, DataPartitionRule>>
private final Map<String, Map<Integer, DataPartitionRule>>
dataPartitionRuleTable;
public PartitionTable() {
this.metadataPartitionTable = new HashMap<>();
- this.mManagerGroupPhysicalNodesMap = new HashMap<>();
+ this.schemaRegionGroupDataNodesMap = new HashMap<>();
this.dataPartitionTable = new HashMap<>();
- this.vsgGroupPhysicalNodesMap = new HashMap<>();
+ this.dataRegionGroupDataNodesMap = new HashMap<>();
this.dataPartitionRuleTable = new HashMap<>();
}
diff --git
a/confignode/src/main/java/org/apache/iotdb/confignode/service/basic/ConfigServiceProvider.java
b/confignode/src/main/java/org/apache/iotdb/confignode/service/basic/ConfigServiceProvider.java
index fca11e8..11a4fa3 100644
---
a/confignode/src/main/java/org/apache/iotdb/confignode/service/basic/ConfigServiceProvider.java
+++
b/confignode/src/main/java/org/apache/iotdb/confignode/service/basic/ConfigServiceProvider.java
@@ -18,28 +18,7 @@
*/
package org.apache.iotdb.confignode.service.basic;
-import org.apache.iotdb.db.exception.StorageEngineException;
-import org.apache.iotdb.db.exception.metadata.StorageGroupNotSetException;
-import org.apache.iotdb.db.exception.query.QueryProcessException;
-import org.apache.iotdb.db.qp.executor.PlanExecutor;
-import org.apache.iotdb.db.qp.physical.PhysicalPlan;
-import org.apache.iotdb.db.query.context.QueryContext;
-import org.apache.iotdb.db.service.basic.ServiceProvider;
+public class ConfigServiceProvider {
-public class ConfigServiceProvider extends ServiceProvider {
- public ConfigServiceProvider(PlanExecutor executor) throws
QueryProcessException {
- super(executor);
- }
-
- @Override
- public QueryContext genQueryContext(
- long queryId, boolean debug, long startTime, String statement, long
timeout) {
- return null;
- }
-
- @Override
- public boolean executeNonQuery(PhysicalPlan plan)
- throws QueryProcessException, StorageGroupNotSetException,
StorageEngineException {
- return false;
- }
+ public ConfigServiceProvider() {}
}
diff --git
a/confignode/src/main/java/org/apache/iotdb/confignode/service/thrift/impl/ConfigServiceImpl.java
b/confignode/src/main/java/org/apache/iotdb/confignode/service/thrift/impl/ConfigServiceImpl.java
index 76c3c1e..0f51939 100644
---
a/confignode/src/main/java/org/apache/iotdb/confignode/service/thrift/impl/ConfigServiceImpl.java
+++
b/confignode/src/main/java/org/apache/iotdb/confignode/service/thrift/impl/ConfigServiceImpl.java
@@ -20,11 +20,6 @@ package org.apache.iotdb.confignode.service.thrift.impl;
import org.apache.iotdb.confignode.manager.ConfigManager;
import org.apache.iotdb.confignode.rpc.thrift.ConfigIService;
-import org.apache.iotdb.confignode.service.basic.ConfigServiceProvider;
-import org.apache.iotdb.db.exception.query.QueryProcessException;
-import org.apache.iotdb.service.rpc.thrift.TSCreateAlignedTimeseriesReq;
-import org.apache.iotdb.service.rpc.thrift.TSCreateMultiTimeseriesReq;
-import org.apache.iotdb.service.rpc.thrift.TSCreateTimeseriesReq;
import org.apache.iotdb.service.rpc.thrift.TSStatus;
import org.apache.thrift.TException;
@@ -34,11 +29,9 @@ import java.util.List;
/** ConfigServiceImpl exposes the interface that interacts with the DataNode */
public class ConfigServiceImpl implements ConfigIService.Iface {
- private ConfigServiceProvider configServiceProvider;
private ConfigManager configManager;
- public ConfigServiceImpl() throws QueryProcessException {
- this.configServiceProvider = new ConfigServiceProvider(null);
+ public ConfigServiceImpl() {
this.configManager = new ConfigManager();
}
@@ -48,27 +41,29 @@ public class ConfigServiceImpl implements
ConfigIService.Iface {
}
@Override
- public TSStatus deleteStorageGroup(long sessionId, List<String>
storageGroups) throws TException {
+ public TSStatus deleteStorageGroup(long sessionId, String storageGroup)
throws TException {
return null;
}
@Override
- public TSStatus createTimeseries(TSCreateTimeseriesReq req) throws
TException {
+ public TSStatus deleteStorageGroups(long sessionId, List<String>
storageGroups)
+ throws TException {
return null;
}
@Override
- public TSStatus createAlignedTimeseries(TSCreateAlignedTimeseriesReq req)
throws TException {
- return null;
+ public int getSchemaPartition(long sessionId, String device) throws
TException {
+ return -1;
}
@Override
- public TSStatus createMultiTimeseries(TSCreateMultiTimeseriesReq req) throws
TException {
+ public List<Integer> getDataPartition(long sessionId, String device,
List<Long> times)
+ throws TException {
return null;
}
@Override
- public TSStatus deleteTimeSeries(long sessionId, List<String> paths) throws
TException {
+ public List<Integer> getLatestDataPartition(long sessionId, String device)
throws TException {
return null;
}
diff --git
a/confignode/src/test/java/org/apache/iotdb/confignode/conf/ConfigNodeDescriptorTest.java
b/confignode/src/test/java/org/apache/iotdb/confignode/conf/ConfigNodeDescriptorTest.java
new file mode 100644
index 0000000..3bc580b
--- /dev/null
+++
b/confignode/src/test/java/org/apache/iotdb/confignode/conf/ConfigNodeDescriptorTest.java
@@ -0,0 +1,76 @@
+/*
+ * 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.conf;
+
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.net.URL;
+
+public class ConfigNodeDescriptorTest {
+
+ private final String confPath = System.getProperty(ConfigNodeConf.CONF_NAME,
null);
+
+ @Before
+ public void init() {
+
org.apache.catalina.webresources.TomcatURLStreamHandlerFactory.getInstance();
+ }
+
+ @After
+ public void clear() {
+ if (confPath != null) {
+ System.setProperty(ConfigNodeConstant.CONFIG_NODE_CONF, confPath);
+ } else {
+ System.clearProperty(ConfigNodeConstant.CONFIG_NODE_CONF);
+ }
+ }
+
+ @Test
+ public void testConfigURLWithFileProtocol() {
+ ConfigNodeDescriptor desc = ConfigNodeDescriptor.getInstance();
+ String pathString = "file:/usr/local/bin";
+
+ System.setProperty(ConfigNodeConstant.CONFIG_NODE_CONF, pathString);
+ URL confURL = desc.getPropsUrl();
+ Assert.assertTrue(confURL.toString().startsWith(pathString));
+ }
+
+ @Test
+ public void testConfigURLWithClasspathProtocol() {
+ ConfigNodeDescriptor desc = ConfigNodeDescriptor.getInstance();
+
+ String pathString = "classpath:/root/path";
+ System.setProperty(ConfigNodeConstant.CONFIG_NODE_CONF, pathString);
+ URL confURL = desc.getPropsUrl();
+ Assert.assertTrue(confURL.toString().startsWith(pathString));
+ }
+
+ @Test
+ public void testConfigURLWithPlainFilePath() {
+ ConfigNodeDescriptor desc = ConfigNodeDescriptor.getInstance();
+ URL path = ConfigNodeConf.class.getResource("/" +
ConfigNodeConf.CONF_NAME);
+ // filePath is a plain path string
+ String filePath = path.getFile();
+ System.setProperty(ConfigNodeConstant.CONFIG_NODE_CONF, filePath);
+ URL confURL = desc.getPropsUrl();
+ Assert.assertEquals(confURL.toString(), path.toString());
+ }
+}
diff --git
a/confignode/src/test/java/org/apache/iotdb/confignode/manager/hash/DeviceGroupHashExecutorTest.java
b/confignode/src/test/java/org/apache/iotdb/confignode/manager/hash/DeviceGroupHashExecutorTest.java
new file mode 100644
index 0000000..7f95cca
--- /dev/null
+++
b/confignode/src/test/java/org/apache/iotdb/confignode/manager/hash/DeviceGroupHashExecutorTest.java
@@ -0,0 +1,95 @@
+/*
+ * 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.hash;
+
+import org.apache.iotdb.confignode.manager.ConfigManager;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Random;
+
+/**
+ * This is a not active test class, which can be used for general index
testing when there is a new
+ * DeviceGroup hash algorithm
+ */
+public class DeviceGroupHashExecutorTest {
+
+ private static final int deviceGroupCount = 10_000;
+ private static final String sg = "root.SGGroup.";
+ private static final int batchCount = 10_000;
+ private static final int batchSize = 10_000;
+ private static final String chars =
+ "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789-";
+
+ private List<String> genBatchDevices() {
+ Random random = new Random();
+ List<String> devices = new ArrayList<>();
+ int fatherLength = random.nextInt(10) + 10;
+ int deviceLength = random.nextInt(5) + 5;
+
+ for (int i = 0; i < batchSize; i++) {
+ StringBuilder curDevice = new StringBuilder(sg);
+ for (int j = 0; j < fatherLength; j++) {
+ curDevice.append(chars.charAt(random.nextInt(chars.length())));
+ }
+ curDevice.append('.');
+ for (int k = 0; k < deviceLength; k++) {
+ curDevice.append(chars.charAt(random.nextInt(chars.length())));
+ }
+ devices.add(curDevice.toString());
+ }
+ return devices;
+ }
+
+ public void GeneralIndexTest() {
+ ConfigManager manager =
+ new ConfigManager(
+ "org.apache.iotdb.confignode.manager.hash.BKDRHashExecutor",
deviceGroupCount);
+ int[] bucket = new int[deviceGroupCount];
+ Arrays.fill(bucket, 0);
+
+ long totalTime = 0;
+ for (int i = 0; i < batchCount; i++) {
+ List<String> devices = genBatchDevices();
+ totalTime -= System.currentTimeMillis();
+ for (String device : devices) {
+ bucket[manager.getDeviceGroupID(device)] += 1;
+ }
+ totalTime += System.currentTimeMillis();
+ }
+
+ Arrays.sort(bucket);
+ int firstNotNull = 0;
+ for (; ; firstNotNull++) {
+ if (bucket[firstNotNull] > 0) {
+ break;
+ }
+ }
+ System.out.println("Empty DeviceGroup count: " + firstNotNull);
+ System.out.println("Minimum DeviceGroup size: " + bucket[firstNotNull]);
+ System.out.println("Maximal DeviceGroup size: " + bucket[deviceGroupCount
- 1]);
+ System.out.println(
+ "Average size of nonempty DeviceGroup: "
+ + (double) (batchCount * batchSize) / (double) (deviceGroupCount -
firstNotNull));
+ System.out.println(
+ "Median size of nonempty DeviceGroup: " + bucket[(deviceGroupCount -
firstNotNull) / 2]);
+ System.out.println("Total time-consuming: " + (double) totalTime / 1000.0
+ "s");
+ }
+}
diff --git a/confignode/src/test/resources/iotdb-confignode.properties
b/confignode/src/test/resources/iotdb-confignode.properties
new file mode 100644
index 0000000..c8d1b46
--- /dev/null
+++ b/confignode/src/test/resources/iotdb-confignode.properties
@@ -0,0 +1,31 @@
+#
+# 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.
+#
+
+####################
+### DeviceGroup Configuration
+####################
+
+# Number of DeviceGroups per StorageGroup
+# Datatype: int
+# device_group_count=10000
+
+# DeviceGroup hash algorithm
+# Datatype: string
+# Currently, BKDR, AP, JS or SDBM hash algorithm is available
+# device_group_hash_algorithm=BKDR
\ No newline at end of file
diff --git a/pom.xml b/pom.xml
index 3413386..9ce80c2 100644
--- a/pom.xml
+++ b/pom.xml
@@ -102,6 +102,7 @@
<module>flink-iotdb-connector</module>
<module>distribution</module>
<module>hive-connector</module>
+ <module>confignode</module>
<module>cluster</module>
<module>cross-tests</module>
<module>zeppelin-interpreter</module>
@@ -110,7 +111,6 @@
<module>client-cpp</module>
<module>metrics</module>
<module>integration</module>
- <module>confignode</module>
<!-- <module>library-udf</module>-->
</modules>
<!-- Properties Management -->
diff --git a/thrift-confignode/src/main/thrift/confignode.thrift
b/thrift-confignode/src/main/thrift/confignode.thrift
index ba71a0e..50b8f4c 100644
--- a/thrift-confignode/src/main/thrift/confignode.thrift
+++ b/thrift-confignode/src/main/thrift/confignode.thrift
@@ -22,17 +22,17 @@ namespace java org.apache.iotdb.confignode.rpc.thrift
namespace py iotdb.thrift.confignode
service ConfigIService {
- rpc.TSStatus setStorageGroup(1:i64 sessionId, 2:string storageGroup)
+ rpc.TSStatus setStorageGroup(1:i64 sessionId, 2:string storageGroup);
- rpc.TSStatus deleteStorageGroup(1:i64 sessionId, 2:list<string>
storageGroups)
+ rpc.TSStatus deleteStorageGroup(1:i64 sessionId, 2:string storageGroup);
- rpc.TSStatus createTimeseries(1:rpc.TSCreateTimeseriesReq req)
+ rpc.TSStatus deleteStorageGroups(1:i64 sessionId, 2:list<string>
storageGroups);
- rpc.TSStatus createAlignedTimeseries(1:rpc.TSCreateAlignedTimeseriesReq req)
+ i32 getSchemaPartition(1:i64 sessionId, 2:string device)
- rpc.TSStatus createMultiTimeseries(1:rpc.TSCreateMultiTimeseriesReq req)
+ list<i32> getDataPartition(1:i64 sessionId, 2:string device, 3:list<i64>
times)
- rpc.TSStatus deleteTimeSeries(1:i64 sessionId, 2:list<string> paths)
+ list<i32> getLatestDataPartition(1:i64 sessionId, 2:string device)
i32 getDeviceGroupID(1:i64 sessionId, 2:string device)
}
\ No newline at end of file
