You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ha...@apache.org on 2021/03/14 03:04:29 UTC
[iotdb] branch HTVector created (now 1daf2a2)
This is an automated email from the ASF dual-hosted git repository.
haonan pushed a change to branch HTVector
in repository https://gitbox.apache.org/repos/asf/iotdb.git.
at 1daf2a2 fix conflicts
This branch includes the following new commits:
new 1daf2a2 fix conflicts
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
[iotdb] 01/01: fix conflicts
Posted by ha...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
haonan pushed a commit to branch HTVector
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 1daf2a2562c6f4bf099b0bf4b01dfc6371f1c943
Author: HTHou <hh...@outlook.com>
AuthorDate: Sun Mar 14 11:04:00 2021 +0800
fix conflicts
---
.github/workflows/client.yml | 30 ++++----
antlr/pom.xml | 19 ++++++
.../antlr4/org/apache/iotdb/db/qp/sql/SqlBase.g4 | 2 +-
client-cpp/README.md | 6 +-
.../resources/conf/iotdb-cluster.properties | 10 ++-
.../java/org/apache/iotdb/cluster/ClusterMain.java | 7 ++
.../cluster/client/async/AsyncClientPool.java | 10 +--
.../iotdb/cluster/client/sync/SyncClientPool.java | 10 +--
.../apache/iotdb/cluster/config/ClusterConfig.java | 15 ++++
.../iotdb/cluster/config/ClusterDescriptor.java | 11 +++
.../iotdb/cluster/coordinator/Coordinator.java | 27 ++++----
.../apache/iotdb/cluster/metadata/CMManager.java | 40 +++++------
.../cluster/query/ClusterPhysicalGenerator.java | 20 +++++-
.../iotdb/cluster/query/ClusterPlanExecutor.java | 4 ++
docs/UserGuide/Operation Manual/Kill Query.md | 2 +-
.../Operation Manual/UDF User Defined Function.md | 2 +-
docs/UserGuide/Server/Cluster Setup.md | 4 +-
docs/zh/UserGuide/Operation Manual/Kill Query.md | 2 +-
.../Operation Manual/UDF User Defined Function.md | 2 +-
docs/zh/UserGuide/Server/Cluster Setup.md | 4 +-
example/client-cpp-example/README.md | 2 +-
hive-connector/pom.xml | 12 ++--
.../resources/conf/iotdb-engine.properties | 2 +-
.../java/org/apache/iotdb/db/conf/IoTDBConfig.java | 10 +--
.../org/apache/iotdb/db/conf/IoTDBDescriptor.java | 4 +-
.../db/engine/compaction/TsFileManagement.java | 8 +++
.../level/LevelCompactionTsFileManagement.java | 31 +++++++++
.../engine/compaction/utils/CompactionUtils.java | 3 +
.../iotdb/db/engine/flush/MemTableFlushTask.java | 7 +-
.../db/engine/storagegroup/TsFileProcessor.java | 4 +-
.../org/apache/iotdb/db/metadata/MManager.java | 10 +++
.../iotdb/db/metadata/template/Template.java | 70 +++++++++++++++++++
.../apache/iotdb/db/qp/executor/PlanExecutor.java | 26 +++++++
.../org/apache/iotdb/db/qp/logical/Operator.java | 4 +-
.../db/qp/physical/crud/CreateTemplatePlan.java | 78 +++++++++++++++++++++
.../db/qp/physical/crud/SetDeviceTemplatePlan.java | 23 +++++++
.../iotdb/db/query/control/QueryTimeManager.java | 2 +-
.../org/apache/iotdb/db/service/TSServiceImpl.java | 79 +++++++++++++++++++++-
.../iotdb/db/integration/IoTDBQueryDemoIT.java | 8 +--
.../java/org/apache/iotdb/session/Session.java | 54 +++++++++++++++
.../apache/iotdb/session/SessionConnection.java | 40 +++++++++++
.../java/org/apache/iotdb/session/SessionUT.java | 47 +++++++++++++
thrift/pom.xml | 19 ++++++
thrift/src/main/thrift/rpc.thrift | 19 ++++++
.../iotdb/tsfile/read/TsFileSequenceReader.java | 15 +++-
...easurementChunkMetadataListMapIteratorTest.java | 65 +++++++++++++++---
.../zeppelin/iotdb/IoTDBInterpreterTest.java | 2 +-
47 files changed, 753 insertions(+), 118 deletions(-)
diff --git a/.github/workflows/client.yml b/.github/workflows/client.yml
index e685c6a..492bf0a 100644
--- a/.github/workflows/client.yml
+++ b/.github/workflows/client.yml
@@ -3,27 +3,27 @@
# CPP compiling is too slow, so let's do it in parallel with testing other modules.
# As there is no Java client, we just use one JDK.
-name: Clients_except_Java CI with Maven
+name: Clients_except_Java CI with Maven
on:
push:
branches:
- master
- - 'rel/*'
+ - "rel/*"
paths:
- - 'client-*/**'
- - 'compile-tools/**'
- - 'thrift/**'
- - 'service-rpc/**'
+ - "client-*/**"
+ - "compile-tools/**"
+ - "thrift/**"
+ - "service-rpc/**"
pull_request:
branches:
- master
- - 'rel/*'
+ - "rel/*"
paths:
- - 'client-*/**'
- - 'compile-tools/**'
- - 'thrift/**'
- - 'service-rpc/**'
+ - "client-*/**"
+ - "compile-tools/**"
+ - "thrift/**"
+ - "service-rpc/**"
# allow manually run the action:
workflow_dispatch:
@@ -45,8 +45,8 @@ jobs:
- name: Install CPP Dependencies (Ubuntu)
if: matrix.os == 'ubuntu-latest'
shell: bash
- run: sudo apt-get install libboost-all-dev
- - name: Install CPP Dependencies (Mac)
+ run: sudo apt-get update && sudo apt-get install libboost-all-dev
+ - name: Install CPP Dependencies (Mac)`
if: matrix.os == 'macos-latest'
shell: bash
run: |
@@ -61,7 +61,7 @@ jobs:
key: client-${{ runner.os }}-m2-${{ hashFiles('**/pom.xml') }}
restore-keys: ${{ runner.os }}-m2-
- name: Test with Maven
- run: mvn -B clean integration-test -P compile-cpp -Dtsfile.test.skip=true -Djdbc.test.skip=true -Diotdb.test.skip=true -Dtest.port.closed=true -pl server,client-cpp,example/client-cpp-example -am
+ run: mvn -B clean integration-test -P compile-cpp -Dtsfile.test.skip=true -Djdbc.test.skip=true -Diotdb.test.skip=true -Dtest.port.closed=true -pl server,client-cpp,example/client-cpp-example -am
build-win:
strategy:
@@ -97,4 +97,4 @@ jobs:
run: cd /d/a/cpp && unzip win_flex_bison.zip && mv win_flex.exe flex.exe && mv win_bison.exe bison.exe && echo 'export PATH=/d/a/cpp:$PATH' >> ~/.bash_profile && source ~/.bash_profile
- name: Test with Maven
shell: bash
- run: source ~/.bash_profile && mvn -B clean integration-test -P compile-cpp -Dboost.include.dir=/d/a/cpp/boost_1_72_0 -Dboost.library.dir=/d/a/cpp/boost_1_72_0/stage/lib -Dtsfile.test.skip=true -Djdbc.test.skip=true -Diotdb.test.skip=true -Dtest.port.closed=true -Denforcer.skip=true -pl server,client-cpp,example/client-cpp-example -am
\ No newline at end of file
+ run: source ~/.bash_profile && mvn -B clean integration-test -P compile-cpp -Dboost.include.dir=/d/a/cpp/boost_1_72_0 -Dboost.library.dir=/d/a/cpp/boost_1_72_0/stage/lib -Dtsfile.test.skip=true -Djdbc.test.skip=true -Diotdb.test.skip=true -Dtest.port.closed=true -Denforcer.skip=true -pl server,client-cpp,example/client-cpp-example -am
diff --git a/antlr/pom.xml b/antlr/pom.xml
index 8ecdb6e..3f1cd44 100644
--- a/antlr/pom.xml
+++ b/antlr/pom.xml
@@ -54,6 +54,25 @@
</execution>
</executions>
</plugin>
+ <plugin>
+ <groupId>org.codehaus.mojo</groupId>
+ <artifactId>build-helper-maven-plugin</artifactId>
+ <version>3.2.0</version>
+ <executions>
+ <execution>
+ <id>add-source</id>
+ <phase>generate-sources</phase>
+ <goals>
+ <goal>add-source</goal>
+ </goals>
+ <configuration>
+ <sources>
+ <source>${project.build.directory}/generated-sources/antlr4</source>
+ </sources>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
</plugins>
</build>
</project>
diff --git a/antlr/src/main/antlr4/org/apache/iotdb/db/qp/sql/SqlBase.g4 b/antlr/src/main/antlr4/org/apache/iotdb/db/qp/sql/SqlBase.g4
index 37d9056..fac9d26 100644
--- a/antlr/src/main/antlr4/org/apache/iotdb/db/qp/sql/SqlBase.g4
+++ b/antlr/src/main/antlr4/org/apache/iotdb/db/qp/sql/SqlBase.g4
@@ -20,7 +20,7 @@
grammar SqlBase;
singleStatement
- : statement EOF
+ : statement (';')? EOF
;
/*
diff --git a/client-cpp/README.md b/client-cpp/README.md
index 3556c54..2800fd4 100644
--- a/client-cpp/README.md
+++ b/client-cpp/README.md
@@ -20,7 +20,7 @@
-->
# Building C++ Client
-To compile cpp client, add "-P client-cpp" option to maven build command.
+To compile cpp client, add "-P compile-cpp" option to maven build command.
The compiling requires the module "compile-tools" to be built first.
For more information, please refer to "compile-tools/README.md".
@@ -28,7 +28,7 @@ For more information, please refer to "compile-tools/README.md".
## Compile and Test:
-`mvn integration-test -P client-cpp -pl client-cpp,server -am -Diotdb.test.skip=true -Dtsfile.test.skip=true -Djdbc.test.skip=true`
+`mvn integration-test -P compile-cpp -pl client-cpp,server -am -Diotdb.test.skip=true -Dtsfile.test.skip=true -Djdbc.test.skip=true`
To compile on Windows, please install Boost first and add following Maven settings:
```
@@ -37,7 +37,7 @@ To compile on Windows, please install Boost first and add following Maven settin
e.g.,
```
-mvn integration-test -P client-cpp -pl client-cpp,server,example/client-cpp-example -am
+mvn integration-test -P compile-cpp -pl client-cpp,server,example/client-cpp-example -am
-D"iotdb.test.skip"=true -D"tsfile.test.skip"=true -D"jdbc.test.skip"=true
-D"boost.include.dir"="D:\boost_1_75_0" -D"boost.library.dir"="D:\boost_1_75_0\stage\lib"
```
diff --git a/cluster/src/assembly/resources/conf/iotdb-cluster.properties b/cluster/src/assembly/resources/conf/iotdb-cluster.properties
index 62aa133..1627d43 100644
--- a/cluster/src/assembly/resources/conf/iotdb-cluster.properties
+++ b/cluster/src/assembly/resources/conf/iotdb-cluster.properties
@@ -157,4 +157,12 @@ max_number_of_logs_per_fetch_on_disk=1000
# When consistency level is set to mid, query will fail if the log lag exceeds max_read_log_lag
# This default value is 1000
-max_read_log_lag=1000
\ No newline at end of file
+max_read_log_lag=1000
+
+# Max number of clients in a ClientPool of a member for one node.
+max_client_pernode_permember_number=1000
+
+# If the number of connections created for a node exceeds `max_client_pernode_permember_number`,
+# we need to wait so much time for other connections to be released until timeout,
+# or a new connection will be created.
+wait_client_timeout_ms=5000
diff --git a/cluster/src/main/java/org/apache/iotdb/cluster/ClusterMain.java b/cluster/src/main/java/org/apache/iotdb/cluster/ClusterMain.java
index db84a86..27aa61b 100644
--- a/cluster/src/main/java/org/apache/iotdb/cluster/ClusterMain.java
+++ b/cluster/src/main/java/org/apache/iotdb/cluster/ClusterMain.java
@@ -30,6 +30,7 @@ import org.apache.iotdb.cluster.rpc.thrift.Node;
import org.apache.iotdb.cluster.server.MetaClusterServer;
import org.apache.iotdb.cluster.server.Response;
import org.apache.iotdb.cluster.utils.ClusterUtils;
+import org.apache.iotdb.db.conf.IoTDBConfigCheck;
import org.apache.iotdb.db.conf.IoTDBConstant;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.exception.StartupException;
@@ -75,6 +76,12 @@ public class ClusterMain {
return;
}
+ try {
+ IoTDBConfigCheck.getInstance().checkConfig();
+ } catch (IOException e) {
+ logger.error("meet error when doing start checking", e);
+ }
+
// init server's configuration first, because the cluster configuration may read settings from
// the server's configuration.
IoTDBDescriptor.getInstance().getConfig().setSyncEnable(false);
diff --git a/cluster/src/main/java/org/apache/iotdb/cluster/client/async/AsyncClientPool.java b/cluster/src/main/java/org/apache/iotdb/cluster/client/async/AsyncClientPool.java
index 1ed9d23..a7441e4 100644
--- a/cluster/src/main/java/org/apache/iotdb/cluster/client/async/AsyncClientPool.java
+++ b/cluster/src/main/java/org/apache/iotdb/cluster/client/async/AsyncClientPool.java
@@ -39,7 +39,7 @@ import java.util.concurrent.ConcurrentHashMap;
public class AsyncClientPool {
private static final Logger logger = LoggerFactory.getLogger(AsyncClientPool.class);
- private static final long WAIT_CLIENT_TIMEOUT_MS = 5 * 1000L;
+ private long waitClientTimeutMS;
private int maxConnectionForEachNode;
private Map<ClusterNode, Deque<AsyncClient>> clientCaches = new ConcurrentHashMap<>();
private Map<ClusterNode, Integer> nodeClientNumMap = new ConcurrentHashMap<>();
@@ -47,6 +47,7 @@ public class AsyncClientPool {
public AsyncClientPool(AsyncClientFactory asyncClientFactory) {
this.asyncClientFactory = asyncClientFactory;
+ this.waitClientTimeutMS = ClusterDescriptor.getInstance().getConfig().getWaitClientTimeoutMS();
this.maxConnectionForEachNode =
ClusterDescriptor.getInstance().getConfig().getMaxClientPerNodePerMember();
}
@@ -122,12 +123,11 @@ public class AsyncClientPool {
long waitStart = System.currentTimeMillis();
while (clientStack.isEmpty()) {
try {
- this.wait(WAIT_CLIENT_TIMEOUT_MS);
- if (clientStack.isEmpty()
- && System.currentTimeMillis() - waitStart >= WAIT_CLIENT_TIMEOUT_MS) {
+ this.wait(waitClientTimeutMS);
+ if (clientStack.isEmpty() && System.currentTimeMillis() - waitStart >= waitClientTimeutMS) {
logger.warn(
"Cannot get an available client after {}ms, create a new one.",
- WAIT_CLIENT_TIMEOUT_MS,
+ waitClientTimeutMS,
asyncClientFactory);
AsyncClient asyncClient = asyncClientFactory.getAsyncClient(clusterNode, this);
nodeClientNumMap.computeIfPresent(clusterNode, (n, oldValue) -> oldValue + 1);
diff --git a/cluster/src/main/java/org/apache/iotdb/cluster/client/sync/SyncClientPool.java b/cluster/src/main/java/org/apache/iotdb/cluster/client/sync/SyncClientPool.java
index 95aba99..758296f 100644
--- a/cluster/src/main/java/org/apache/iotdb/cluster/client/sync/SyncClientPool.java
+++ b/cluster/src/main/java/org/apache/iotdb/cluster/client/sync/SyncClientPool.java
@@ -38,7 +38,7 @@ import java.util.concurrent.ConcurrentHashMap;
public class SyncClientPool {
private static final Logger logger = LoggerFactory.getLogger(SyncClientPool.class);
- private static final long WAIT_CLIENT_TIMEOUT_MS = 5 * 1000L;
+ private long waitClientTimeoutMS;
private int maxConnectionForEachNode;
private Map<ClusterNode, Deque<Client>> clientCaches = new ConcurrentHashMap<>();
private Map<ClusterNode, Integer> nodeClientNumMap = new ConcurrentHashMap<>();
@@ -46,6 +46,7 @@ public class SyncClientPool {
public SyncClientPool(SyncClientFactory syncClientFactory) {
this.syncClientFactory = syncClientFactory;
+ this.waitClientTimeoutMS = ClusterDescriptor.getInstance().getConfig().getWaitClientTimeoutMS();
this.maxConnectionForEachNode =
ClusterDescriptor.getInstance().getConfig().getMaxClientPerNodePerMember();
}
@@ -112,12 +113,11 @@ public class SyncClientPool {
long waitStart = System.currentTimeMillis();
while (clientStack.isEmpty()) {
try {
- this.wait(WAIT_CLIENT_TIMEOUT_MS);
+ this.wait(waitClientTimeoutMS);
if (clientStack.isEmpty()
- && System.currentTimeMillis() - waitStart >= WAIT_CLIENT_TIMEOUT_MS) {
+ && System.currentTimeMillis() - waitStart >= waitClientTimeoutMS) {
logger.warn(
- "Cannot get an available client after {}ms, create a new one",
- WAIT_CLIENT_TIMEOUT_MS);
+ "Cannot get an available client after {}ms, create a new one", waitClientTimeoutMS);
Client client = syncClientFactory.getSyncClient(clusterNode, this);
nodeClientNumMap.computeIfPresent(clusterNode, (n, oldValue) -> oldValue + 1);
return client;
diff --git a/cluster/src/main/java/org/apache/iotdb/cluster/config/ClusterConfig.java b/cluster/src/main/java/org/apache/iotdb/cluster/config/ClusterConfig.java
index 6fe99dc..08f7608 100644
--- a/cluster/src/main/java/org/apache/iotdb/cluster/config/ClusterConfig.java
+++ b/cluster/src/main/java/org/apache/iotdb/cluster/config/ClusterConfig.java
@@ -72,6 +72,13 @@ public class ClusterConfig {
private int maxClientPerNodePerMember = 1000;
/**
+ * If the number of connections created for a node exceeds `max_client_pernode_permember_number`,
+ * we need to wait so much time for other connections to be released until timeout, or a new
+ * connection will be created.
+ */
+ private long waitClientTimeoutMS = 5 * 1000L;
+
+ /**
* ClientPool will have so many selector threads (TAsyncClientManager) to distribute to its
* clients.
*/
@@ -441,4 +448,12 @@ public class ClusterConfig {
public void setOpenServerRpcPort(boolean openServerRpcPort) {
this.openServerRpcPort = openServerRpcPort;
}
+
+ public long getWaitClientTimeoutMS() {
+ return waitClientTimeoutMS;
+ }
+
+ public void setWaitClientTimeoutMS(long waitClientTimeoutMS) {
+ this.waitClientTimeoutMS = waitClientTimeoutMS;
+ }
}
diff --git a/cluster/src/main/java/org/apache/iotdb/cluster/config/ClusterDescriptor.java b/cluster/src/main/java/org/apache/iotdb/cluster/config/ClusterDescriptor.java
index 631662f..d6bbf82 100644
--- a/cluster/src/main/java/org/apache/iotdb/cluster/config/ClusterDescriptor.java
+++ b/cluster/src/main/java/org/apache/iotdb/cluster/config/ClusterDescriptor.java
@@ -282,6 +282,17 @@ public class ClusterDescriptor {
Long.parseLong(
properties.getProperty("max_read_log_lag", String.valueOf(config.getMaxReadLogLag()))));
+ config.setMaxClientPerNodePerMember(
+ Integer.parseInt(
+ properties.getProperty(
+ "max_client_pernode_permember_number",
+ String.valueOf(config.getMaxClientPerNodePerMember()))));
+
+ config.setWaitClientTimeoutMS(
+ Long.parseLong(
+ properties.getProperty(
+ "wait_client_timeout_ms", String.valueOf(config.getWaitClientTimeoutMS()))));
+
String consistencyLevel = properties.getProperty("consistency_level");
if (consistencyLevel != null) {
config.setConsistencyLevel(ConsistencyLevel.getConsistencyLevel(consistencyLevel));
diff --git a/cluster/src/main/java/org/apache/iotdb/cluster/coordinator/Coordinator.java b/cluster/src/main/java/org/apache/iotdb/cluster/coordinator/Coordinator.java
index 820831c..e8be569 100644
--- a/cluster/src/main/java/org/apache/iotdb/cluster/coordinator/Coordinator.java
+++ b/cluster/src/main/java/org/apache/iotdb/cluster/coordinator/Coordinator.java
@@ -159,31 +159,32 @@ public class Coordinator {
* nodes.
*/
private TSStatus processNonPartitionedDataPlan(PhysicalPlan plan) {
- if (plan instanceof DeleteTimeSeriesPlan || plan instanceof DeletePlan) {
- try {
+ try {
+ if (plan instanceof DeleteTimeSeriesPlan || plan instanceof DeletePlan) {
// as delete related plans may have abstract paths (paths with wildcards), we convert
// them to full paths so the executor nodes will not need to query the metadata holders,
// eliminating the risk that when they are querying the metadata holders, the timeseries
// has already been deleted
((CMManager) IoTDB.metaManager).convertToFullPaths(plan);
- } catch (PathNotExistException e) {
- if (plan.getPaths().isEmpty()) {
- // only reports an error when there is no matching path
- return StatusUtils.getStatus(StatusUtils.TIMESERIES_NOT_EXIST_ERROR, e.getMessage());
- }
+ } else {
+ // function convertToFullPaths has already sync leader
+ metaGroupMember.syncLeaderWithConsistencyCheck(true);
+ }
+ } catch (PathNotExistException e) {
+ if (plan.getPaths().isEmpty()) {
+ // only reports an error when there is no matching path
+ return StatusUtils.getStatus(StatusUtils.TIMESERIES_NOT_EXIST_ERROR, e.getMessage());
}
- }
- try {
- metaGroupMember.syncLeaderWithConsistencyCheck(true);
- List<PartitionGroup> globalGroups = metaGroupMember.getPartitionTable().getGlobalGroups();
- logger.debug("Forwarding global data plan {} to {} groups", plan, globalGroups.size());
- return forwardPlan(globalGroups, plan);
} catch (CheckConsistencyException e) {
logger.debug(
"Forwarding global data plan {} to meta leader {}", plan, metaGroupMember.getLeader());
metaGroupMember.waitLeader();
return metaGroupMember.forwardPlan(plan, metaGroupMember.getLeader(), null);
}
+
+ List<PartitionGroup> globalGroups = metaGroupMember.getPartitionTable().getGlobalGroups();
+ logger.debug("Forwarding global data plan {} to {} groups", plan, globalGroups.size());
+ return forwardPlan(globalGroups, plan);
}
/**
diff --git a/cluster/src/main/java/org/apache/iotdb/cluster/metadata/CMManager.java b/cluster/src/main/java/org/apache/iotdb/cluster/metadata/CMManager.java
index 223e383..279eeae 100644
--- a/cluster/src/main/java/org/apache/iotdb/cluster/metadata/CMManager.java
+++ b/cluster/src/main/java/org/apache/iotdb/cluster/metadata/CMManager.java
@@ -152,6 +152,19 @@ public class CMManager extends MManager {
return CMManager.MManagerHolder.INSTANCE;
}
+ /**
+ * sync meta leader to get the newest partition table and storage groups.
+ *
+ * @throws MetadataException throws MetadataException if necessary
+ */
+ public void syncMetaLeader() throws MetadataException {
+ try {
+ metaGroupMember.syncLeaderWithConsistencyCheck(false);
+ } catch (CheckConsistencyException e) {
+ throw new MetadataException(e);
+ }
+ }
+
@Override
public String deleteTimeseries(PartialPath prefixPath) throws MetadataException {
cacheLock.writeLock().lock();
@@ -1020,12 +1033,6 @@ public class CMManager extends MManager {
* @return all paths after removing wildcards in the path
*/
public Set<PartialPath> getMatchedDevices(PartialPath originPath) throws MetadataException {
- // make sure this node knows all storage groups
- try {
- metaGroupMember.syncLeaderWithConsistencyCheck(false);
- } catch (CheckConsistencyException e) {
- throw new MetadataException(e);
- }
// get all storage groups this path may belong to
// the key is the storage group name and the value is the path to be queried with storage group
// added, e.g:
@@ -1283,12 +1290,7 @@ public class CMManager extends MManager {
@Override
public Pair<List<PartialPath>, Integer> getAllTimeseriesPathWithAlias(
PartialPath prefixPath, int limit, int offset) throws MetadataException {
- // make sure this node knows all storage groups
- try {
- metaGroupMember.syncLeaderWithConsistencyCheck(false);
- } catch (CheckConsistencyException e) {
- throw new MetadataException(e);
- }
+
// get all storage groups this path may belong to
// the key is the storage group name and the value is the path to be queried with storage group
// added, e.g:
@@ -1321,12 +1323,6 @@ public class CMManager extends MManager {
* @return all paths after removing wildcards in the path
*/
public List<PartialPath> getMatchedPaths(PartialPath originPath) throws MetadataException {
- // make sure this node knows all storage groups
- try {
- metaGroupMember.syncLeaderWithConsistencyCheck(false);
- } catch (CheckConsistencyException e) {
- throw new MetadataException(e);
- }
// get all storage groups this path may belong to
// the key is the storage group name and the value is the path to be queried with storage group
// added, e.g:
@@ -1429,7 +1425,11 @@ public class CMManager extends MManager {
* Replace partial paths (paths not containing measurements), and abstract paths (paths containing
* wildcards) with full paths.
*/
- public void convertToFullPaths(PhysicalPlan plan) throws PathNotExistException {
+ public void convertToFullPaths(PhysicalPlan plan)
+ throws PathNotExistException, CheckConsistencyException {
+ // make sure this node knows all storage groups
+ metaGroupMember.syncLeaderWithConsistencyCheck(false);
+
Pair<List<PartialPath>, List<PartialPath>> getMatchedPathsRet =
getMatchedPaths(plan.getPaths());
List<PartialPath> fullPaths = getMatchedPathsRet.left;
@@ -1761,6 +1761,8 @@ public class CMManager extends MManager {
if (withAlias) {
alias = new ArrayList<>();
}
+ // make sure this node knows all storage groups
+ syncMetaLeader();
if (withAlias) {
for (String path : paths) {
diff --git a/cluster/src/main/java/org/apache/iotdb/cluster/query/ClusterPhysicalGenerator.java b/cluster/src/main/java/org/apache/iotdb/cluster/query/ClusterPhysicalGenerator.java
index ccfa7d7..a159dee 100644
--- a/cluster/src/main/java/org/apache/iotdb/cluster/query/ClusterPhysicalGenerator.java
+++ b/cluster/src/main/java/org/apache/iotdb/cluster/query/ClusterPhysicalGenerator.java
@@ -25,6 +25,8 @@ import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.exception.metadata.MetadataException;
import org.apache.iotdb.db.exception.query.QueryProcessException;
import org.apache.iotdb.db.metadata.PartialPath;
+import org.apache.iotdb.db.qp.logical.Operator;
+import org.apache.iotdb.db.qp.logical.crud.SFWOperator;
import org.apache.iotdb.db.qp.logical.sys.LoadConfigurationOperator.LoadConfigurationOperatorType;
import org.apache.iotdb.db.qp.physical.PhysicalPlan;
import org.apache.iotdb.db.qp.physical.sys.LoadConfigurationPlan;
@@ -67,12 +69,26 @@ public class ClusterPhysicalGenerator extends PhysicalGenerator {
@Override
protected List<PartialPath> getMatchedTimeseries(PartialPath path) throws MetadataException {
- return ((CMManager) IoTDB.metaManager).getMatchedPaths(path);
+ return getCMManager().getMatchedPaths(path);
}
@Override
protected Set<PartialPath> getMatchedDevices(PartialPath path) throws MetadataException {
- return ((CMManager) IoTDB.metaManager).getMatchedDevices(path);
+ return getCMManager().getMatchedDevices(path);
+ }
+
+ @Override
+ public PhysicalPlan transformToPhysicalPlan(Operator operator, int fetchSize)
+ throws QueryProcessException {
+ // update storage groups before parsing query plans
+ if (operator instanceof SFWOperator) {
+ try {
+ getCMManager().syncMetaLeader();
+ } catch (MetadataException e) {
+ throw new QueryProcessException(e);
+ }
+ }
+ return super.transformToPhysicalPlan(operator, fetchSize);
}
@Override
diff --git a/cluster/src/main/java/org/apache/iotdb/cluster/query/ClusterPlanExecutor.java b/cluster/src/main/java/org/apache/iotdb/cluster/query/ClusterPlanExecutor.java
index 6b78916..fd5f638 100644
--- a/cluster/src/main/java/org/apache/iotdb/cluster/query/ClusterPlanExecutor.java
+++ b/cluster/src/main/java/org/apache/iotdb/cluster/query/ClusterPlanExecutor.java
@@ -54,6 +54,7 @@ import org.apache.iotdb.db.query.context.QueryContext;
import org.apache.iotdb.db.query.dataset.AlignByDeviceDataSet;
import org.apache.iotdb.db.query.executor.IQueryRouter;
import org.apache.iotdb.db.service.IoTDB;
+import org.apache.iotdb.db.utils.TestOnly;
import org.apache.iotdb.tsfile.exception.filter.QueryFilterOptimizationException;
import org.apache.iotdb.tsfile.read.query.dataset.QueryDataSet;
@@ -118,6 +119,7 @@ public class ClusterPlanExecutor extends PlanExecutor {
}
@Override
+ @TestOnly
protected List<PartialPath> getPathsName(PartialPath path) throws MetadataException {
return ((CMManager) IoTDB.metaManager).getMatchedPaths(path);
}
@@ -283,6 +285,8 @@ public class ClusterPlanExecutor extends PlanExecutor {
@Override
protected Set<PartialPath> getDevices(PartialPath path) throws MetadataException {
+ // make sure this node knows all storage groups
+ ((CMManager) IoTDB.metaManager).syncMetaLeader();
return ((CMManager) IoTDB.metaManager).getMatchedDevices(path);
}
diff --git a/docs/UserGuide/Operation Manual/Kill Query.md b/docs/UserGuide/Operation Manual/Kill Query.md
index 13c4368..9b4e543 100644
--- a/docs/UserGuide/Operation Manual/Kill Query.md
+++ b/docs/UserGuide/Operation Manual/Kill Query.md
@@ -30,7 +30,7 @@ For queries that take too long to execute, IoTDB will forcibly interrupt the que
-The default timeout of the system is 60000 ms，which can be customized in the configuration file through the `query_time_threshold` parameter.
+The default timeout of the system is 60000 ms，which can be customized in the configuration file through the `query_timeout_threshold` parameter.
If you use JDBC or Session, we also support setting a timeout for a single query（Unit: ms）：
diff --git a/docs/UserGuide/Operation Manual/UDF User Defined Function.md b/docs/UserGuide/Operation Manual/UDF User Defined Function.md
index 76a20e8..c90cf3c 100644
--- a/docs/UserGuide/Operation Manual/UDF User Defined Function.md
+++ b/docs/UserGuide/Operation Manual/UDF User Defined Function.md
@@ -571,7 +571,7 @@ When you have prepared the UDF source code, test cases, and instructions, you ar
## Known Implementation UDF Libraries
-+ [TsClean](https://thulab.github.io/tsclean-iotdb), a UDF library about data quality, including data quality indicator calculation, value filling, value repairing, etc.
++ [IoTDB-Quality](https://thulab.github.io/iotdb-quality), a UDF library about data quality, including data profiling, data quality evalution and data repairing, etc.
## Q&A
diff --git a/docs/UserGuide/Server/Cluster Setup.md b/docs/UserGuide/Server/Cluster Setup.md
index b8d484c..2af632c 100644
--- a/docs/UserGuide/Server/Cluster Setup.md
+++ b/docs/UserGuide/Server/Cluster Setup.md
@@ -45,9 +45,9 @@ To start the service of one of the nodes, you need to execute the following comm
## Example of pseudo-distributed scaffolding for 3 nodes and 3 replicas
```
-mvn clean package -pl cluster -am -Dmaven.test.skip=true
+mvn clean package -DskipTests
chmod -R 777 ./cluster/target/
-nohup ./cluster/target/cluster-0.12.0-SNAPSHOT/sbin/start-node.sh ./cluster/target/test-classes/node1conf/ >/dev/null 2>&1 &
+nohup ./cluster/target/cluster-0.12.0-SNAPSHOT/sbin/start-node.sh ./cluster/target/test-classes/node1conf/ >/dev/null 2>&1 &
nohup ./cluster/target/cluster-0.12.0-SNAPSHOT/sbin/start-node.sh ./cluster/target/test-classes/node2conf/ >/dev/null 2>&1 &
nohup ./cluster/target/cluster-0.12.0-SNAPSHOT/sbin/start-node.sh ./cluster/target/test-classes/node3conf/ >/dev/null 2>&1 &
```
diff --git a/docs/zh/UserGuide/Operation Manual/Kill Query.md b/docs/zh/UserGuide/Operation Manual/Kill Query.md
index 8755f05..13917ec 100644
--- a/docs/zh/UserGuide/Operation Manual/Kill Query.md
+++ b/docs/zh/UserGuide/Operation Manual/Kill Query.md
@@ -31,7 +31,7 @@
-系统默认的超时时间为 60000 ms，可以在配置文件中通过 `query_time_threshold` 参数进行自定义配置。
+系统默认的超时时间为 60000 ms，可以在配置文件中通过 `query_timeout_threshold` 参数进行自定义配置。
如果您使用 JDBC 或 Session，还支持对单个查询设置超时时间（单位为 ms）：
diff --git a/docs/zh/UserGuide/Operation Manual/UDF User Defined Function.md b/docs/zh/UserGuide/Operation Manual/UDF User Defined Function.md
index b52dd2f..0b4e260 100644
--- a/docs/zh/UserGuide/Operation Manual/UDF User Defined Function.md
+++ b/docs/zh/UserGuide/Operation Manual/UDF User Defined Function.md
@@ -574,7 +574,7 @@ SHOW FUNCTIONS
## 已知的UDF库实现
-+ [TsClean](https://thulab.github.io/tsclean-iotdb)，一个关于数据质量的UDF库实现，包括时序数据的质量指标计算、数值填补、数值修复等一系列函数。
++ [IoTDB-Quality](https://thulab.github.io/iotdb-quality)，一个关于数据质量的UDF库实现，包括数据画像、数据质量评估与修复等一系列函数。
## Q&A
diff --git a/docs/zh/UserGuide/Server/Cluster Setup.md b/docs/zh/UserGuide/Server/Cluster Setup.md
index fe89ef2..686f7eb 100644
--- a/docs/zh/UserGuide/Server/Cluster Setup.md
+++ b/docs/zh/UserGuide/Server/Cluster Setup.md
@@ -41,9 +41,9 @@ __集群模式目前是测试版！请谨慎在生产环境中使用。__
## 3节点3副本伪分布式搭建示例
```
-mvn clean package -pl cluster -am -Dmaven.test.skip=true
+mvn clean package -DskipTests
chmod -R 777 ./cluster/target/
-nohup ./cluster/target/cluster-0.12.0-SNAPSHOT/sbin/start-node.sh ./cluster/target/test-classes/node1conf/ >/dev/null 2>&1 &
+nohup ./cluster/target/cluster-0.12.0-SNAPSHOT/sbin/start-node.sh ./cluster/target/test-classes/node1conf/ >/dev/null 2>&1 &
nohup ./cluster/target/cluster-0.12.0-SNAPSHOT/sbin/start-node.sh ./cluster/target/test-classes/node2conf/ >/dev/null 2>&1 &
nohup ./cluster/target/cluster-0.12.0-SNAPSHOT/sbin/start-node.sh ./cluster/target/test-classes/node3conf/ >/dev/null 2>&1 &
```
diff --git a/example/client-cpp-example/README.md b/example/client-cpp-example/README.md
index 07d16b7..33292d8 100644
--- a/example/client-cpp-example/README.md
+++ b/example/client-cpp-example/README.md
@@ -26,7 +26,7 @@
using maven to build this example project:
* cd the root path of the whole project
-* run `mvn package -DskipTests -P client-cpp -pl example/client-cpp-example -am`
+* run `mvn package -DskipTests -P compile-cpp -pl example/client-cpp-example -am`
* cd example/client-cpp-example/target
You can find some files to form a complete project:
diff --git a/hive-connector/pom.xml b/hive-connector/pom.xml
index 7a0355d..5afd8da 100644
--- a/hive-connector/pom.xml
+++ b/hive-connector/pom.xml
@@ -168,17 +168,13 @@
</dependency>
</dependencies>
</dependencyManagement>
- <!-- org.pentaho:pentaho-aggdesigner-algorithm is in repo.spring.io and we have to claim
- to use https-->
- <!-- Please note the notice in https://spring.io/blog/2020/10/29/notice-of-permissions-changes-to-repo-spring-io-fall-and-winter-2020
- Anonymous access using /libs-snapshot or /libs-milestone in the pom.xml, or with
- these configured in a remote repository, should replace them with /snapshot and /milestone,
- respectively.-->
+ <!-- org.pentaho:pentaho-aggdesigner-algorithm is removed from repo.spring.io and we have to claim
+ to use https://public.nexus.pentaho.org or http://conjars.org/repo -->
<repositories>
<repository>
<id>for_pentaho</id>
- <name>spring.io</name>
- <url>https://repo.spring.io/milestone</url>
+ <name>public.nexus.pentaho.org</name>
+ <url>https://public.nexus.pentaho.org/repository/proxy-public-3rd-party-release</url>
<layout>default</layout>
<releases>
<enabled>true</enabled>
diff --git a/server/src/assembly/resources/conf/iotdb-engine.properties b/server/src/assembly/resources/conf/iotdb-engine.properties
index eb527ae..8c4c6fb 100644
--- a/server/src/assembly/resources/conf/iotdb-engine.properties
+++ b/server/src/assembly/resources/conf/iotdb-engine.properties
@@ -380,7 +380,7 @@ compaction_thread_num=10
merge_write_throughput_mb_per_sec=8
# The max executing time of query. unit: ms
-query_time_threshold=60000
+query_timeout_threshold=60000
####################
### Metadata Cache Configuration
diff --git a/server/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java b/server/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
index 146533e..44addbc 100644
--- a/server/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
+++ b/server/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
@@ -363,7 +363,7 @@ public class IoTDBConfig {
private long cacheFileReaderClearPeriod = 100000;
/** the max executing time of query in ms. */
- private int queryTimeThreshold = 60000;
+ private int queryTimeoutThreshold = 60000;
/** Replace implementation class of JDBC service */
private String rpcImplClassName = TSServiceImpl.class.getName();
@@ -1119,12 +1119,12 @@ public class IoTDBConfig {
this.cacheFileReaderClearPeriod = cacheFileReaderClearPeriod;
}
- public int getQueryTimeThreshold() {
- return queryTimeThreshold;
+ public int getQueryTimeoutThreshold() {
+ return queryTimeoutThreshold;
}
- public void setQueryTimeThreshold(int queryTimeThreshold) {
- this.queryTimeThreshold = queryTimeThreshold;
+ public void setQueryTimeoutThreshold(int queryTimeoutThreshold) {
+ this.queryTimeoutThreshold = queryTimeoutThreshold;
}
public boolean isReadOnly() {
diff --git a/server/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java b/server/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java
index 86b6134..2ef9347 100644
--- a/server/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java
+++ b/server/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java
@@ -342,10 +342,10 @@ public class IoTDBDescriptor {
"unseq_file_num_in_each_level",
Integer.toString(conf.getUnseqFileNumInEachLevel()))));
- conf.setQueryTimeThreshold(
+ conf.setQueryTimeoutThreshold(
Integer.parseInt(
properties.getProperty(
- "query_time_threshold", Integer.toString(conf.getQueryTimeThreshold()))));
+ "query_timeout_threshold", Integer.toString(conf.getQueryTimeoutThreshold()))));
conf.setSyncEnable(
Boolean.parseBoolean(
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/compaction/TsFileManagement.java b/server/src/main/java/org/apache/iotdb/db/engine/compaction/TsFileManagement.java
index 1b18011..b176365 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/compaction/TsFileManagement.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/compaction/TsFileManagement.java
@@ -34,6 +34,7 @@ import org.apache.iotdb.db.engine.modification.ModificationFile;
import org.apache.iotdb.db.engine.storagegroup.StorageGroupProcessor.CloseCompactionMergeCallBack;
import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
import org.apache.iotdb.db.exception.MergeException;
+import org.apache.iotdb.tsfile.fileSystem.FSFactoryProducer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -47,6 +48,7 @@ import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import static org.apache.iotdb.db.conf.IoTDBConstant.FILE_NAME_SEPARATOR;
+import static org.apache.iotdb.db.engine.merge.task.MergeTask.MERGE_SUFFIX;
import static org.apache.iotdb.db.engine.storagegroup.StorageGroupProcessor.MERGING_MODIFICATION_FILE_NAME;
import static org.apache.iotdb.tsfile.common.constant.TsFileConstant.TSFILE_SUFFIX;
@@ -383,6 +385,12 @@ public abstract class TsFileManagement {
doubleWriteLock(seqFile);
try {
+ // if meet error(like file not found) in merge task, the .merge file may not be deleted
+ File mergedFile =
+ FSFactoryProducer.getFSFactory().getFile(seqFile.getTsFilePath() + MERGE_SUFFIX);
+ if (mergedFile.exists()) {
+ mergedFile.delete();
+ }
updateMergeModification(seqFile);
if (i == seqFiles.size() - 1) {
// FIXME if there is an exception, the the modification file will be not closed.
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/compaction/level/LevelCompactionTsFileManagement.java b/server/src/main/java/org/apache/iotdb/db/engine/compaction/level/LevelCompactionTsFileManagement.java
index 91d572e..8dfbd2b 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/compaction/level/LevelCompactionTsFileManagement.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/compaction/level/LevelCompactionTsFileManagement.java
@@ -664,6 +664,7 @@ public class LevelCompactionTsFileManagement extends TsFileManagement {
}
}
} catch (Exception e) {
+ restoreCompaction();
logger.error("Error occurred in Compaction Merge thread", e);
} finally {
isSeqMerging = false;
@@ -771,6 +772,36 @@ public class LevelCompactionTsFileManagement extends TsFileManagement {
throw new IOException();
}
+ /** restore the files back to the status before the compaction task is submitted */
+ private void restoreCompaction() {
+ File logFile =
+ FSFactoryProducer.getFSFactory()
+ .getFile(storageGroupDir, storageGroupName + COMPACTION_LOG_NAME);
+ try {
+ if (logFile.exists()) {
+ CompactionLogAnalyzer logAnalyzer = new CompactionLogAnalyzer(logFile);
+ logAnalyzer.analyze();
+ String targetFilePath = logAnalyzer.getTargetFile();
+ if (targetFilePath != null) {
+ File targetFile = new File(targetFilePath);
+ if (targetFile.exists()) {
+ targetFile.delete();
+ }
+ }
+ }
+ } catch (IOException e) {
+ logger.error("restore compaction failed", e);
+ } finally {
+ if (logFile.exists()) {
+ try {
+ Files.delete(logFile.toPath());
+ } catch (IOException e) {
+ logger.error("delete compaction log file error ", e);
+ }
+ }
+ }
+ }
+
@TestOnly
public Map<Long, List<SortedSet<TsFileResource>>> getSequenceTsFileResources() {
return sequenceTsFileResources;
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/compaction/utils/CompactionUtils.java b/server/src/main/java/org/apache/iotdb/db/engine/compaction/utils/CompactionUtils.java
index bfdd52a..059b07b 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/compaction/utils/CompactionUtils.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/compaction/utils/CompactionUtils.java
@@ -271,6 +271,9 @@ public class CompactionUtils {
for (TsFileResource tsFileResource : tsFileResources) {
TsFileSequenceReader reader =
buildReaderFromTsFileResource(tsFileResource, tsFileSequenceReaderMap, storageGroup);
+ if (reader == null) {
+ throw new IOException();
+ }
Iterator<Map<String, List<ChunkMetadata>>> iterator =
reader.getMeasurementChunkMetadataListMapIterator(device);
chunkMetadataListIteratorCache.put(reader, iterator);
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/flush/MemTableFlushTask.java b/server/src/main/java/org/apache/iotdb/db/engine/flush/MemTableFlushTask.java
index c60ec2e..3413082 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/flush/MemTableFlushTask.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/flush/MemTableFlushTask.java
@@ -28,7 +28,6 @@ import org.apache.iotdb.db.rescon.SystemInfo;
import org.apache.iotdb.db.utils.datastructure.TVList;
import org.apache.iotdb.db.utils.datastructure.VectorTVList;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
-import org.apache.iotdb.tsfile.utils.Binary;
import org.apache.iotdb.tsfile.utils.Pair;
import org.apache.iotdb.tsfile.write.chunk.ChunkWriterImpl;
import org.apache.iotdb.tsfile.write.chunk.IChunkWriter;
@@ -45,8 +44,6 @@ import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;
-import static org.apache.iotdb.db.rescon.PrimitiveArrayManager.ARRAY_SIZE;
-
public class MemTableFlushTask {
private static final Logger LOGGER = LoggerFactory.getLogger(MemTableFlushTask.class);
@@ -221,7 +218,9 @@ public class MemTableFlushTask {
break;
default:
LOGGER.error(
- "Storage group {} does not support data type: {}", storageGroup, dataType);
+ "Storage group {} does not support data type: {}",
+ storageGroup,
+ dataType);
break;
}
}
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileProcessor.java b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileProcessor.java
index 6a4d9ab..886df75 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileProcessor.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileProcessor.java
@@ -57,7 +57,6 @@ import org.apache.iotdb.rpc.RpcUtils;
import org.apache.iotdb.rpc.TSStatusCode;
import org.apache.iotdb.service.rpc.thrift.TSStatus;
import org.apache.iotdb.tsfile.file.metadata.ChunkMetadata;
-import org.apache.iotdb.tsfile.file.metadata.VectorChunkMetadata;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.file.metadata.enums.TSEncoding;
import org.apache.iotdb.tsfile.read.common.TimeRange;
@@ -294,8 +293,7 @@ public class TsFileProcessor {
if (insertRowPlan.getDataTypes()[i] == TSDataType.VECTOR) {
// TODO: insertRowPlan
// chunkMetadataIncrement += VectorChunkMetadata.calculateRamSize(insertRowPlan....);
- }
- else {
+ } else {
chunkMetadataIncrement +=
ChunkMetadata.calculateRamSize(
insertRowPlan.getMeasurements()[i], insertRowPlan.getDataTypes()[i]);
diff --git a/server/src/main/java/org/apache/iotdb/db/metadata/MManager.java b/server/src/main/java/org/apache/iotdb/db/metadata/MManager.java
index 76703a8..2ca42e1 100644
--- a/server/src/main/java/org/apache/iotdb/db/metadata/MManager.java
+++ b/server/src/main/java/org/apache/iotdb/db/metadata/MManager.java
@@ -38,12 +38,15 @@ import org.apache.iotdb.db.metadata.logfile.MLogWriter;
import org.apache.iotdb.db.metadata.mnode.MNode;
import org.apache.iotdb.db.metadata.mnode.MeasurementMNode;
import org.apache.iotdb.db.metadata.mnode.StorageGroupMNode;
+import org.apache.iotdb.db.metadata.template.Template;
import org.apache.iotdb.db.monitor.MonitorConstants;
import org.apache.iotdb.db.qp.constant.SQLConstant;
import org.apache.iotdb.db.qp.physical.PhysicalPlan;
+import org.apache.iotdb.db.qp.physical.crud.CreateTemplatePlan;
import org.apache.iotdb.db.qp.physical.crud.InsertPlan;
import org.apache.iotdb.db.qp.physical.crud.InsertRowPlan;
import org.apache.iotdb.db.qp.physical.crud.InsertTabletPlan;
+import org.apache.iotdb.db.qp.physical.crud.SetDeviceTemplatePlan;
import org.apache.iotdb.db.qp.physical.sys.ChangeAliasPlan;
import org.apache.iotdb.db.qp.physical.sys.ChangeTagOffsetPlan;
import org.apache.iotdb.db.qp.physical.sys.CreateAlignedTimeSeriesPlan;
@@ -161,6 +164,9 @@ public class MManager {
private static final int ESTIMATED_SERIES_SIZE = config.getEstimatedSeriesSize();
+ // template name -> template
+ private Map<String, Template> templateHashMap = new ConcurrentHashMap<>();
+
private static class MManagerHolder {
private MManagerHolder() {
@@ -1968,4 +1974,8 @@ public class MManager {
boolean satisfy(String storageGroup);
}
+
+ public void createDeviceTemplate(CreateTemplatePlan plan) throws MetadataException {}
+
+ public void setDeviceTemplate(SetDeviceTemplatePlan plan) throws MetadataException {}
}
diff --git a/server/src/main/java/org/apache/iotdb/db/metadata/template/Template.java b/server/src/main/java/org/apache/iotdb/db/metadata/template/Template.java
new file mode 100644
index 0000000..9aef180
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/metadata/template/Template.java
@@ -0,0 +1,70 @@
+/*
+ * 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.db.metadata.template;
+
+import org.apache.iotdb.db.qp.physical.crud.CreateTemplatePlan;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSEncoding;
+import org.apache.iotdb.tsfile.write.schema.IMeasurementSchema;
+import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
+import org.apache.iotdb.tsfile.write.schema.VectorMeasurementSchema;
+
+import java.util.HashMap;
+import java.util.Map;
+
+public class Template {
+ String name;
+ Map<String, IMeasurementSchema> schemaMap = new HashMap<>();
+
+ public Template(CreateTemplatePlan plan) {
+ name = plan.getName();
+
+ // put measurement into a map
+ for (int i = 0; i < plan.getMeasurements().size(); i++) {
+ IMeasurementSchema curSchema;
+ // vector
+ if (plan.getMeasurements().get(i).size() > 1) {
+ curSchema =
+ new VectorMeasurementSchema(
+ (String[]) plan.getMeasurements().get(i).toArray(),
+ (TSDataType[]) plan.getDataTypes().get(i).toArray(),
+ (TSEncoding[]) plan.getEncodings().get(i).toArray(),
+ plan.getCompressors().get(i));
+ }
+ // normal measurement
+ else {
+ curSchema =
+ new MeasurementSchema(
+ plan.getMeasurements().get(i).get(0),
+ plan.getDataTypes().get(i).get(0),
+ plan.getEncodings().get(i).get(0),
+ plan.getCompressors().get(i));
+ }
+
+ for (String path : plan.getMeasurements().get(i)) {
+ if (schemaMap.containsKey(path)) {
+ throw new IllegalArgumentException(
+ "Duplicate measurement name in create template plan. Name is :" + path);
+ }
+
+ schemaMap.put(path, curSchema);
+ }
+ }
+ }
+}
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/executor/PlanExecutor.java b/server/src/main/java/org/apache/iotdb/db/qp/executor/PlanExecutor.java
index 4c008ca..9f0a77f 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/executor/PlanExecutor.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/executor/PlanExecutor.java
@@ -56,6 +56,7 @@ import org.apache.iotdb.db.qp.logical.sys.AuthorOperator.AuthorType;
import org.apache.iotdb.db.qp.physical.PhysicalPlan;
import org.apache.iotdb.db.qp.physical.crud.AggregationPlan;
import org.apache.iotdb.db.qp.physical.crud.AlignByDevicePlan;
+import org.apache.iotdb.db.qp.physical.crud.CreateTemplatePlan;
import org.apache.iotdb.db.qp.physical.crud.DeletePartitionPlan;
import org.apache.iotdb.db.qp.physical.crud.DeletePlan;
import org.apache.iotdb.db.qp.physical.crud.FillQueryPlan;
@@ -71,6 +72,7 @@ import org.apache.iotdb.db.qp.physical.crud.LastQueryPlan;
import org.apache.iotdb.db.qp.physical.crud.QueryIndexPlan;
import org.apache.iotdb.db.qp.physical.crud.QueryPlan;
import org.apache.iotdb.db.qp.physical.crud.RawDataQueryPlan;
+import org.apache.iotdb.db.qp.physical.crud.SetDeviceTemplatePlan;
import org.apache.iotdb.db.qp.physical.crud.UDTFPlan;
import org.apache.iotdb.db.qp.physical.sys.AlterTimeSeriesPlan;
import org.apache.iotdb.db.qp.physical.sys.AuthorPlan;
@@ -338,12 +340,36 @@ public class PlanExecutor implements IPlanExecutor {
throw new QueryProcessException(e.getMessage());
}
return true;
+ case CREATE_TEMPLATE:
+ return createDeviceTemplate((CreateTemplatePlan) plan);
+ case SET_DEVICE_TEMPLATE:
+ return setDeviceTemplate((SetDeviceTemplatePlan) plan);
default:
throw new UnsupportedOperationException(
String.format("operation %s is not supported", plan.getOperatorType()));
}
}
+ private boolean createDeviceTemplate(CreateTemplatePlan createTemplatePlan)
+ throws QueryProcessException {
+ try {
+ IoTDB.metaManager.createDeviceTemplate(createTemplatePlan);
+ } catch (MetadataException e) {
+ throw new QueryProcessException(e);
+ }
+ return true;
+ }
+
+ private boolean setDeviceTemplate(SetDeviceTemplatePlan setDeviceTemplatePlan)
+ throws QueryProcessException {
+ try {
+ IoTDB.metaManager.setDeviceTemplate(setDeviceTemplatePlan);
+ } catch (MetadataException e) {
+ throw new QueryProcessException(e);
+ }
+ return true;
+ }
+
private boolean operateCreateFunction(CreateFunctionPlan plan) throws UDFRegistrationException {
UDFRegistrationService.getInstance()
.register(plan.getUdfName(), plan.getClassName(), plan.isTemporary(), true);
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/logical/Operator.java b/server/src/main/java/org/apache/iotdb/db/qp/logical/Operator.java
index 372ac46..bc2a902 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/logical/Operator.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/logical/Operator.java
@@ -151,6 +151,8 @@ public abstract class Operator {
CREATE_TRIGGER,
DROP_TRIGGER,
START_TRIGGER,
- STOP_TRIGGER
+ STOP_TRIGGER,
+ CREATE_TEMPLATE,
+ SET_DEVICE_TEMPLATE,
}
}
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/CreateTemplatePlan.java b/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/CreateTemplatePlan.java
new file mode 100644
index 0000000..69a1015
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/CreateTemplatePlan.java
@@ -0,0 +1,78 @@
+package org.apache.iotdb.db.qp.physical.crud;
+
+import org.apache.iotdb.db.metadata.PartialPath;
+import org.apache.iotdb.db.qp.logical.Operator.OperatorType;
+import org.apache.iotdb.db.qp.physical.PhysicalPlan;
+import org.apache.iotdb.tsfile.file.metadata.enums.CompressionType;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSEncoding;
+
+import java.util.List;
+
+public class CreateTemplatePlan extends PhysicalPlan {
+
+ String name;
+ List<List<String>> measurements;
+ List<List<TSDataType>> dataTypes;
+ List<List<TSEncoding>> encodings;
+ List<CompressionType> compressors;
+
+ public String getName() {
+ return name;
+ }
+
+ public void setName(String name) {
+ this.name = name;
+ }
+
+ public List<List<String>> getMeasurements() {
+ return measurements;
+ }
+
+ public void setMeasurements(List<List<String>> measurements) {
+ this.measurements = measurements;
+ }
+
+ public List<List<TSDataType>> getDataTypes() {
+ return dataTypes;
+ }
+
+ public void setDataTypes(List<List<TSDataType>> dataTypes) {
+ this.dataTypes = dataTypes;
+ }
+
+ public List<List<TSEncoding>> getEncodings() {
+ return encodings;
+ }
+
+ public void setEncodings(List<List<TSEncoding>> encodings) {
+ this.encodings = encodings;
+ }
+
+ public List<CompressionType> getCompressors() {
+ return compressors;
+ }
+
+ public void setCompressors(List<CompressionType> compressors) {
+ this.compressors = compressors;
+ }
+
+ public CreateTemplatePlan(
+ String name,
+ List<List<String>> measurements,
+ List<List<TSDataType>> dataTypes,
+ List<List<TSEncoding>> encodings,
+ List<CompressionType> compressors) {
+ super(false, OperatorType.CREATE_TEMPLATE);
+ this.name = name;
+ this.measurements = measurements;
+ this.dataTypes = dataTypes;
+ this.encodings = encodings;
+ this.compressors = compressors;
+ }
+
+ @Override
+ public List<PartialPath> getPaths() {
+ return null;
+ }
+}
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/SetDeviceTemplatePlan.java b/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/SetDeviceTemplatePlan.java
new file mode 100644
index 0000000..08475cc
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/SetDeviceTemplatePlan.java
@@ -0,0 +1,23 @@
+package org.apache.iotdb.db.qp.physical.crud;
+
+import org.apache.iotdb.db.metadata.PartialPath;
+import org.apache.iotdb.db.qp.logical.Operator.OperatorType;
+import org.apache.iotdb.db.qp.physical.PhysicalPlan;
+
+import java.util.List;
+
+public class SetDeviceTemplatePlan extends PhysicalPlan {
+ String templateName;
+ String prefixPath;
+
+ public SetDeviceTemplatePlan(String templateName, String prefixPath) {
+ super(false, OperatorType.SET_DEVICE_TEMPLATE);
+ this.templateName = templateName;
+ this.prefixPath = prefixPath;
+ }
+
+ @Override
+ public List<PartialPath> getPaths() {
+ return null;
+ }
+}
diff --git a/server/src/main/java/org/apache/iotdb/db/query/control/QueryTimeManager.java b/server/src/main/java/org/apache/iotdb/db/query/control/QueryTimeManager.java
index 6f2cebd..59e646f 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/control/QueryTimeManager.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/control/QueryTimeManager.java
@@ -61,7 +61,7 @@ public class QueryTimeManager implements IService {
}
public void registerQuery(long queryId, long startTime, String sql, long timeout) {
- final long finalTimeout = timeout == 0 ? config.getQueryTimeThreshold() : timeout;
+ final long finalTimeout = timeout == 0 ? config.getQueryTimeoutThreshold() : timeout;
queryInfoMap.put(queryId, new QueryInfo(startTime, sql));
// submit a scheduled task to judge whether query is still running after timeout
ScheduledFuture<?> scheduledFuture =
diff --git a/server/src/main/java/org/apache/iotdb/db/service/TSServiceImpl.java b/server/src/main/java/org/apache/iotdb/db/service/TSServiceImpl.java
index 33feaf7..876d154 100644
--- a/server/src/main/java/org/apache/iotdb/db/service/TSServiceImpl.java
+++ b/server/src/main/java/org/apache/iotdb/db/service/TSServiceImpl.java
@@ -50,6 +50,7 @@ import org.apache.iotdb.db.qp.physical.PhysicalPlan;
import org.apache.iotdb.db.qp.physical.crud.AggregationPlan;
import org.apache.iotdb.db.qp.physical.crud.AlignByDevicePlan;
import org.apache.iotdb.db.qp.physical.crud.AlignByDevicePlan.MeasurementType;
+import org.apache.iotdb.db.qp.physical.crud.CreateTemplatePlan;
import org.apache.iotdb.db.qp.physical.crud.DeletePlan;
import org.apache.iotdb.db.qp.physical.crud.GroupByTimePlan;
import org.apache.iotdb.db.qp.physical.crud.InsertMultiTabletPlan;
@@ -60,6 +61,7 @@ import org.apache.iotdb.db.qp.physical.crud.InsertTabletPlan;
import org.apache.iotdb.db.qp.physical.crud.LastQueryPlan;
import org.apache.iotdb.db.qp.physical.crud.QueryPlan;
import org.apache.iotdb.db.qp.physical.crud.RawDataQueryPlan;
+import org.apache.iotdb.db.qp.physical.crud.SetDeviceTemplatePlan;
import org.apache.iotdb.db.qp.physical.crud.UDFPlan;
import org.apache.iotdb.db.qp.physical.crud.UDTFPlan;
import org.apache.iotdb.db.qp.physical.sys.AuthorPlan;
@@ -91,6 +93,7 @@ import org.apache.iotdb.service.rpc.thrift.TSCancelOperationReq;
import org.apache.iotdb.service.rpc.thrift.TSCloseOperationReq;
import org.apache.iotdb.service.rpc.thrift.TSCloseSessionReq;
import org.apache.iotdb.service.rpc.thrift.TSCreateAlignedTimeseriesReq;
+import org.apache.iotdb.service.rpc.thrift.TSCreateDeviceTemplateReq;
import org.apache.iotdb.service.rpc.thrift.TSCreateMultiTimeseriesReq;
import org.apache.iotdb.service.rpc.thrift.TSCreateTimeseriesReq;
import org.apache.iotdb.service.rpc.thrift.TSDeleteDataReq;
@@ -116,6 +119,7 @@ import org.apache.iotdb.service.rpc.thrift.TSProtocolVersion;
import org.apache.iotdb.service.rpc.thrift.TSQueryDataSet;
import org.apache.iotdb.service.rpc.thrift.TSQueryNonAlignDataSet;
import org.apache.iotdb.service.rpc.thrift.TSRawDataQueryReq;
+import org.apache.iotdb.service.rpc.thrift.TSSetDeviceTemplateReq;
import org.apache.iotdb.service.rpc.thrift.TSSetTimeZoneReq;
import org.apache.iotdb.service.rpc.thrift.TSStatus;
import org.apache.iotdb.tsfile.exception.filter.QueryFilterOptimizationException;
@@ -582,7 +586,7 @@ public class TSServiceImpl implements TSIService.Iface, ServerContext {
req.statementId,
physicalPlan,
req.fetchSize,
- config.getQueryTimeThreshold(),
+ config.getQueryTimeoutThreshold(),
sessionIdUsernameMap.get(req.getSessionId()))
: RpcUtils.getTSExecuteStatementResp(
TSStatusCode.EXECUTE_STATEMENT_ERROR, "Statement is not a query statement.");
@@ -1747,6 +1751,79 @@ public class TSServiceImpl implements TSIService.Iface, ServerContext {
return statementId;
}
+ @Override
+ public TSStatus createDeviceTemplate(TSCreateDeviceTemplateReq req) throws TException {
+ try {
+ if (!checkLogin(req.getSessionId())) {
+ return RpcUtils.getStatus(TSStatusCode.NOT_LOGIN_ERROR);
+ }
+
+ if (AUDIT_LOGGER.isDebugEnabled()) {
+ AUDIT_LOGGER.debug(
+ "Session-{} create device template {}.{}.{}.{}.{}",
+ currSessionId.get(),
+ req.getName(),
+ req.getMeasurements(),
+ req.getDataTypes(),
+ req.getEncodings(),
+ req.getCompressors());
+ }
+
+ List<List<TSDataType>> dataTypes = new ArrayList<>();
+ for (List<Integer> list : req.dataTypes) {
+ List<TSDataType> dataTypesList = new ArrayList<>();
+ for (int dataType : list) {
+ dataTypesList.add(TSDataType.values()[dataType]);
+ }
+ dataTypes.add(dataTypesList);
+ }
+
+ List<List<TSEncoding>> encodings = new ArrayList<>();
+ for (List<Integer> list : req.dataTypes) {
+ List<TSEncoding> encodingsList = new ArrayList<>();
+ for (int encoding : list) {
+ encodingsList.add(TSEncoding.values()[encoding]);
+ }
+ encodings.add(encodingsList);
+ }
+
+ List<CompressionType> compressionTypes = new ArrayList<>();
+ for (int compressType : req.getCompressors()) {
+ compressionTypes.add(CompressionType.values()[compressType]);
+ }
+
+ CreateTemplatePlan plan =
+ new CreateTemplatePlan(
+ req.getName(), req.getMeasurements(), dataTypes, encodings, compressionTypes);
+
+ TSStatus status = checkAuthority(plan, req.getSessionId());
+ return status != null ? status : executeNonQueryPlan(plan);
+ } catch (Exception e) {
+ return onNPEOrUnexpectedException(
+ e, "creating aligned timeseries", TSStatusCode.EXECUTE_STATEMENT_ERROR);
+ }
+ }
+
+ @Override
+ public TSStatus setDeviceTemplate(TSSetDeviceTemplateReq req) throws TException {
+ if (!checkLogin(req.getSessionId())) {
+ return RpcUtils.getStatus(TSStatusCode.NOT_LOGIN_ERROR);
+ }
+
+ if (AUDIT_LOGGER.isDebugEnabled()) {
+ AUDIT_LOGGER.debug(
+ "Session-{} set device template {}.{}",
+ currSessionId.get(),
+ req.getTemplate(),
+ req.getPrefixPath());
+ }
+
+ SetDeviceTemplatePlan plan = new SetDeviceTemplatePlan(req.template, req.prefixPath);
+
+ TSStatus status = checkAuthority(plan, req.getSessionId());
+ return status != null ? status : executeNonQueryPlan(plan);
+ }
+
private TSStatus checkAuthority(PhysicalPlan plan, long sessionId) {
List<PartialPath> paths = plan.getPaths();
try {
diff --git a/server/src/test/java/org/apache/iotdb/db/integration/IoTDBQueryDemoIT.java b/server/src/test/java/org/apache/iotdb/db/integration/IoTDBQueryDemoIT.java
index 208478c..ba247fc 100644
--- a/server/src/test/java/org/apache/iotdb/db/integration/IoTDBQueryDemoIT.java
+++ b/server/src/test/java/org/apache/iotdb/db/integration/IoTDBQueryDemoIT.java
@@ -63,10 +63,10 @@ public class IoTDBQueryDemoIT {
"insert into root.ln.wf01.wt01(timestamp,temperature) values(1509465780000,20.182663)",
"insert into root.ln.wf01.wt01(timestamp,temperature) values(1509465840000,21.125198)",
"insert into root.ln.wf01.wt01(timestamp,temperature) values(1509465900000,22.720892)",
- "insert into root.ln.wf01.wt01(timestamp,temperature) values(1509465960000,20.71)",
- "insert into root.ln.wf01.wt01(timestamp,temperature) values(1509466020000,21.451046)",
- "insert into root.ln.wf01.wt01(timestamp,temperature) values(1509466080000,22.57987)",
- "insert into root.ln.wf01.wt01(timestamp,temperature) values(1509466140000,20.98177)",
+ "insert into root.ln.wf01.wt01(timestamp,temperature) values(1509465960000,20.71);",
+ "insert into root.ln.wf01.wt01(timestamp,temperature) values(1509466020000,21.451046);",
+ "insert into root.ln.wf01.wt01(timestamp,temperature) values(1509466080000,22.57987);",
+ "insert into root.ln.wf01.wt01(timestamp,temperature) values(1509466140000,20.98177);",
"create timeseries root.ln.wf02.wt02.hardware with datatype=TEXT,encoding=PLAIN",
"insert into root.ln.wf02.wt02(timestamp,hardware) values(1509465600000,\"v2\")",
"insert into root.ln.wf02.wt02(timestamp,hardware) values(1509465660000,\"v2\")",
diff --git a/session/src/main/java/org/apache/iotdb/session/Session.java b/session/src/main/java/org/apache/iotdb/session/Session.java
index b5b793f..a43918b 100644
--- a/session/src/main/java/org/apache/iotdb/session/Session.java
+++ b/session/src/main/java/org/apache/iotdb/session/Session.java
@@ -24,6 +24,7 @@ import org.apache.iotdb.rpc.RedirectException;
import org.apache.iotdb.rpc.StatementExecutionException;
import org.apache.iotdb.service.rpc.thrift.EndPoint;
import org.apache.iotdb.service.rpc.thrift.TSCreateAlignedTimeseriesReq;
+import org.apache.iotdb.service.rpc.thrift.TSCreateDeviceTemplateReq;
import org.apache.iotdb.service.rpc.thrift.TSCreateMultiTimeseriesReq;
import org.apache.iotdb.service.rpc.thrift.TSCreateTimeseriesReq;
import org.apache.iotdb.service.rpc.thrift.TSDeleteDataReq;
@@ -35,6 +36,7 @@ import org.apache.iotdb.service.rpc.thrift.TSInsertStringRecordsReq;
import org.apache.iotdb.service.rpc.thrift.TSInsertTabletReq;
import org.apache.iotdb.service.rpc.thrift.TSInsertTabletsReq;
import org.apache.iotdb.service.rpc.thrift.TSProtocolVersion;
+import org.apache.iotdb.service.rpc.thrift.TSSetDeviceTemplateReq;
import org.apache.iotdb.tsfile.common.conf.TSFileConfig;
import org.apache.iotdb.tsfile.exception.write.UnSupportedDataTypeException;
import org.apache.iotdb.tsfile.file.metadata.enums.CompressionType;
@@ -1487,4 +1489,56 @@ public class Session {
throw new UnSupportedDataTypeException(MSG_UNSUPPORTED_DATA_TYPE + dataType);
}
}
+
+ public void setDeviceTemplate(String templateName, String prefixPath)
+ throws IoTDBConnectionException, StatementExecutionException {
+ TSSetDeviceTemplateReq request = getTSSetDeviceTemplateReq(templateName, prefixPath);
+ defaultSessionConnection.setDeviceTemplate(request);
+ }
+
+ public void createDeviceTemplate(
+ String name,
+ List<List<String>> measurements,
+ List<List<TSDataType>> dataTypes,
+ List<List<TSEncoding>> encodings,
+ List<CompressionType> compressors)
+ throws IoTDBConnectionException, StatementExecutionException {
+ TSCreateDeviceTemplateReq request =
+ getTSCreateDeviceTemplateReq(name, measurements, dataTypes, encodings, compressors);
+ defaultSessionConnection.createDeviceTemplate(request);
+ }
+
+ private TSSetDeviceTemplateReq getTSSetDeviceTemplateReq(String templateName, String prefixPath) {
+ TSSetDeviceTemplateReq request = new TSSetDeviceTemplateReq();
+ request.setTemplate(templateName);
+ request.setPrefixPath(prefixPath);
+ return request;
+ }
+
+ private TSCreateDeviceTemplateReq getTSCreateDeviceTemplateReq(
+ String name,
+ List<List<String>> measurements,
+ List<List<TSDataType>> dataTypes,
+ List<List<TSEncoding>> encodings,
+ List<CompressionType> compressors) {
+ TSCreateDeviceTemplateReq request = new TSCreateDeviceTemplateReq();
+ request.setName(name);
+ request.setMeasurements(measurements);
+
+ List<List<Integer>> requestType = new ArrayList<>();
+ for (List<TSDataType> typesList : dataTypes) {
+ requestType.add(typesList.stream().map(TSDataType::ordinal).collect(Collectors.toList()));
+ }
+ request.setDataTypes(requestType);
+
+ List<List<Integer>> requestEncoding = new ArrayList<>();
+ for (List<TSEncoding> encodingList : encodings) {
+ requestEncoding.add(
+ encodingList.stream().map(TSEncoding::ordinal).collect(Collectors.toList()));
+ }
+ request.setEncodings(requestEncoding);
+ request.setCompressors(
+ compressors.stream().map(CompressionType::ordinal).collect(Collectors.toList()));
+ return request;
+ }
}
diff --git a/session/src/main/java/org/apache/iotdb/session/SessionConnection.java b/session/src/main/java/org/apache/iotdb/session/SessionConnection.java
index 32e3701..c11b719 100644
--- a/session/src/main/java/org/apache/iotdb/session/SessionConnection.java
+++ b/session/src/main/java/org/apache/iotdb/session/SessionConnection.java
@@ -27,6 +27,7 @@ import org.apache.iotdb.rpc.StatementExecutionException;
import org.apache.iotdb.service.rpc.thrift.EndPoint;
import org.apache.iotdb.service.rpc.thrift.TSCloseSessionReq;
import org.apache.iotdb.service.rpc.thrift.TSCreateAlignedTimeseriesReq;
+import org.apache.iotdb.service.rpc.thrift.TSCreateDeviceTemplateReq;
import org.apache.iotdb.service.rpc.thrift.TSCreateMultiTimeseriesReq;
import org.apache.iotdb.service.rpc.thrift.TSCreateTimeseriesReq;
import org.apache.iotdb.service.rpc.thrift.TSDeleteDataReq;
@@ -43,6 +44,7 @@ import org.apache.iotdb.service.rpc.thrift.TSInsertTabletsReq;
import org.apache.iotdb.service.rpc.thrift.TSOpenSessionReq;
import org.apache.iotdb.service.rpc.thrift.TSOpenSessionResp;
import org.apache.iotdb.service.rpc.thrift.TSRawDataQueryReq;
+import org.apache.iotdb.service.rpc.thrift.TSSetDeviceTemplateReq;
import org.apache.iotdb.service.rpc.thrift.TSSetTimeZoneReq;
import org.apache.iotdb.service.rpc.thrift.TSStatus;
@@ -680,4 +682,42 @@ public class SessionConnection {
}
return flag;
}
+
+ protected void createDeviceTemplate(TSCreateDeviceTemplateReq request)
+ throws IoTDBConnectionException, StatementExecutionException {
+ request.setSessionId(sessionId);
+ try {
+ RpcUtils.verifySuccess(client.createDeviceTemplate(request));
+ } catch (TException e) {
+ if (reconnect()) {
+ try {
+ request.setSessionId(sessionId);
+ RpcUtils.verifySuccess(client.createDeviceTemplate(request));
+ } catch (TException tException) {
+ throw new IoTDBConnectionException(tException);
+ }
+ } else {
+ throw new IoTDBConnectionException(MSG_RECONNECTION_FAIL);
+ }
+ }
+ }
+
+ protected void setDeviceTemplate(TSSetDeviceTemplateReq request)
+ throws IoTDBConnectionException, StatementExecutionException {
+ request.setSessionId(sessionId);
+ try {
+ RpcUtils.verifySuccess(client.setDeviceTemplate(request));
+ } catch (TException e) {
+ if (reconnect()) {
+ try {
+ request.setSessionId(sessionId);
+ RpcUtils.verifySuccess(client.setDeviceTemplate(request));
+ } catch (TException tException) {
+ throw new IoTDBConnectionException(tException);
+ }
+ } else {
+ throw new IoTDBConnectionException(MSG_RECONNECTION_FAIL);
+ }
+ }
+ }
}
diff --git a/session/src/test/java/org/apache/iotdb/session/SessionUT.java b/session/src/test/java/org/apache/iotdb/session/SessionUT.java
index 26d0ddd..e3db07d 100644
--- a/session/src/test/java/org/apache/iotdb/session/SessionUT.java
+++ b/session/src/test/java/org/apache/iotdb/session/SessionUT.java
@@ -37,6 +37,7 @@ import org.junit.Test;
import java.time.ZoneId;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.List;
import static org.junit.Assert.assertArrayEquals;
@@ -199,4 +200,50 @@ public class SessionUT {
session.setTimeout(60000);
Assert.assertEquals(60000, session.getTimeout());
}
+
+ @Test
+ public void setDeviceTemplate() throws IoTDBConnectionException, StatementExecutionException {
+ session = new Session("127.0.0.1", 6667, "root", "root", ZoneId.of("+05:00"));
+ session.open();
+
+ session.setDeviceTemplate("template1", "root.sg.1");
+ }
+
+ @Test
+ public void createDeviceTemplate() throws IoTDBConnectionException, StatementExecutionException {
+ session = new Session("127.0.0.1", 6667, "root", "root", ZoneId.of("+05:00"));
+ session.open();
+
+ List<List<String>> measurementList = new ArrayList<>();
+ measurementList.add(Collections.singletonList("s11"));
+ List<String> measurements = new ArrayList<>();
+ for (int i = 0; i < 10; i++) {
+ measurements.add("s" + i);
+ }
+ measurementList.add(measurements);
+
+ List<List<TSDataType>> dataTypeList = new ArrayList<>();
+ dataTypeList.add(Collections.singletonList(TSDataType.INT64));
+ List<TSDataType> dataTypes = new ArrayList<>();
+ for (int i = 0; i < 10; i++) {
+ dataTypes.add(TSDataType.INT64);
+ }
+ dataTypeList.add(dataTypes);
+
+ List<List<TSEncoding>> encodingList = new ArrayList<>();
+ encodingList.add(Collections.singletonList(TSEncoding.RLE));
+ List<TSEncoding> encodings = new ArrayList<>();
+ for (int i = 0; i < 10; i++) {
+ encodings.add(TSEncoding.RLE);
+ }
+ encodingList.add(encodings);
+
+ List<CompressionType> compressionTypes = new ArrayList<>();
+ for (int i = 0; i < 11; i++) {
+ compressionTypes.add(CompressionType.SNAPPY);
+ }
+
+ session.createDeviceTemplate(
+ "template1", measurementList, dataTypeList, encodingList, compressionTypes);
+ }
}
diff --git a/thrift/pom.xml b/thrift/pom.xml
index 989bf85..7a0e2c6 100644
--- a/thrift/pom.xml
+++ b/thrift/pom.xml
@@ -74,6 +74,25 @@
</archive>
</configuration>
</plugin>
+ <plugin>
+ <groupId>org.codehaus.mojo</groupId>
+ <artifactId>build-helper-maven-plugin</artifactId>
+ <version>3.2.0</version>
+ <executions>
+ <execution>
+ <id>add-source</id>
+ <phase>generate-sources</phase>
+ <goals>
+ <goal>add-source</goal>
+ </goals>
+ <configuration>
+ <sources>
+ <source>${project.build.directory}/generated-sources/thrift</source>
+ </sources>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
</plugins>
</build>
</project>
diff --git a/thrift/src/main/thrift/rpc.thrift b/thrift/src/main/thrift/rpc.thrift
index 44498d9..698da2e 100644
--- a/thrift/src/main/thrift/rpc.thrift
+++ b/thrift/src/main/thrift/rpc.thrift
@@ -306,6 +306,21 @@ struct ServerProperties {
3: required string timestampPrecision;
}
+struct TSSetDeviceTemplateReq {
+ 1: required i64 sessionId
+ 2: required string template
+ 3: required string prefixPath
+}
+
+struct TSCreateDeviceTemplateReq {
+ 1: required i64 sessionId
+ 2: required string name
+ 3: required list<list<string>> measurements
+ 4: required list<list<i32>> dataTypes
+ 5: required list<list<i32>> encodings
+ 6: required list<i32> compressors
+}
+
service TSIService {
TSOpenSessionResp openSession(1:TSOpenSessionReq req);
@@ -378,4 +393,8 @@ service TSIService {
TSExecuteStatementResp executeRawDataQuery(1:TSRawDataQueryReq req);
i64 requestStatementId(1:i64 sessionId);
+
+ TSStatus createDeviceTemplate(1:TSCreateDeviceTemplateReq req);
+
+ TSStatus setDeviceTemplate(1:TSSetDeviceTemplateReq req);
}
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/read/TsFileSequenceReader.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/read/TsFileSequenceReader.java
index e1013fd..9717c31 100644
--- a/tsfile/src/main/java/org/apache/iotdb/tsfile/read/TsFileSequenceReader.java
+++ b/tsfile/src/main/java/org/apache/iotdb/tsfile/read/TsFileSequenceReader.java
@@ -65,6 +65,7 @@ import java.util.Comparator;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
+import java.util.LinkedHashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
@@ -1261,6 +1262,13 @@ public class TsFileSequenceReader implements AutoCloseable {
return maxPlanIndex;
}
+ /**
+ * @return An iterator of linked hashmaps ( measurement -> chunk metadata list ). When traversing
+ * the linked hashmap, you will get chunk metadata lists according to the lexicographic order
+ * of the measurements. The first measurement of the linked hashmap of each iteration is
+ * always larger than the last measurement of the linked hashmap of the previous iteration in
+ * lexicographic order.
+ */
public Iterator<Map<String, List<ChunkMetadata>>> getMeasurementChunkMetadataListMapIterator(
String device) throws IOException {
readFileMetadata();
@@ -1278,7 +1286,7 @@ public class TsFileSequenceReader implements AutoCloseable {
}
@Override
- public Map<String, List<ChunkMetadata>> next() {
+ public LinkedHashMap<String, List<ChunkMetadata>> next() {
throw new NoSuchElementException();
}
};
@@ -1296,12 +1304,13 @@ public class TsFileSequenceReader implements AutoCloseable {
}
@Override
- public Map<String, List<ChunkMetadata>> next() {
+ public LinkedHashMap<String, List<ChunkMetadata>> next() {
if (!hasNext()) {
throw new NoSuchElementException();
}
Pair<Long, Long> startEndPair = queue.remove();
- Map<String, List<ChunkMetadata>> measurementChunkMetadataList = new HashMap<>();
+ LinkedHashMap<String, List<ChunkMetadata>> measurementChunkMetadataList =
+ new LinkedHashMap<>();
try {
List<TimeseriesMetadata> timeseriesMetadataList = new ArrayList<>();
ByteBuffer nextBuffer = readData(startEndPair.left, startEndPair.right);
diff --git a/tsfile/src/test/java/org/apache/iotdb/tsfile/read/MeasurementChunkMetadataListMapIteratorTest.java b/tsfile/src/test/java/org/apache/iotdb/tsfile/read/MeasurementChunkMetadataListMapIteratorTest.java
index c7afdf2..7c72a0d 100644
--- a/tsfile/src/test/java/org/apache/iotdb/tsfile/read/MeasurementChunkMetadataListMapIteratorTest.java
+++ b/tsfile/src/test/java/org/apache/iotdb/tsfile/read/MeasurementChunkMetadataListMapIteratorTest.java
@@ -60,45 +60,65 @@ public class MeasurementChunkMetadataListMapIteratorTest {
@Test
public void test0() throws IOException {
- test(1, 1);
+ testCorrectness(1, 1);
+ testSequentiality(1, 1);
}
@Test
public void test1() throws IOException {
- test(1, 10);
+ testCorrectness(1, 10);
+ testSequentiality(1, 10);
}
@Test
public void test2() throws IOException {
- test(2, 1);
+ testCorrectness(2, 1);
+ testSequentiality(2, 1);
}
@Test
public void test3() throws IOException {
- test(2, 2);
+ testCorrectness(2, 2);
+ testSequentiality(2, 2);
}
@Test
public void test4() throws IOException {
- test(2, 100);
+ testCorrectness(2, 100);
+ testSequentiality(2, 100);
}
@Test
public void test5() throws IOException {
- test(50, 2);
+ testCorrectness(50, 2);
+ testSequentiality(50, 2);
}
@Test
public void test6() throws IOException {
- test(50, 50);
+ testCorrectness(50, 50);
+ testSequentiality(50, 50);
}
@Test
public void test7() throws IOException {
- test(50, 100);
+ testCorrectness(50, 100);
+ testSequentiality(50, 100);
}
- public void test(int deviceNum, int measurementNum) throws IOException {
+ @Test
+ public void test8() throws IOException {
+ testCorrectness(33, 733);
+ testSequentiality(33, 733);
+ }
+
+ @Test
+ public void test9() throws IOException {
+ testCorrectness(733, 33);
+ testSequentiality(733, 33);
+ }
+
+ public void testCorrectness(int deviceNum, int measurementNum) throws IOException {
FileGenerator.generateFile(10000, deviceNum, measurementNum);
try (TsFileSequenceReader fileReader = new TsFileSequenceReader(FILE_PATH)) {
@@ -131,14 +151,14 @@ public class MeasurementChunkMetadataListMapIteratorTest {
}
}
- check(expected, actual);
+ checkCorrectness(expected, actual);
}
}
FileGenerator.after();
}
- private void check(
+ private void checkCorrectness(
Map<String, List<IChunkMetadata>> expected, Map<String, List<IChunkMetadata>> actual) {
Assert.assertEquals(expected.keySet(), actual.keySet());
for (String measurement : expected.keySet()) {
@@ -152,4 +172,27 @@ public class MeasurementChunkMetadataListMapIteratorTest {
}
}
}
+
+ public void testSequentiality(int deviceNum, int measurementNum) throws IOException {
+ FileGenerator.generateFile(10000, deviceNum, measurementNum);
+
+ try (TsFileSequenceReader fileReader = new TsFileSequenceReader(FILE_PATH)) {
+ for (String device : fileReader.getAllDevices()) {
+ Iterator<Map<String, List<ChunkMetadata>>> iterator =
+ fileReader.getMeasurementChunkMetadataListMapIterator(device);
+
+ String lastMeasurement = null;
+ while (iterator.hasNext()) {
+ for (String measurement : iterator.next().keySet()) {
+ if (lastMeasurement != null) {
+ Assert.assertTrue(lastMeasurement.compareTo(measurement) < 0);
+ }
+ lastMeasurement = measurement;
+ }
+ }
+ }
+ }
+
+ FileGenerator.after();
+ }
}
diff --git a/zeppelin-interpreter/src/test/java/org/apache/zeppelin/iotdb/IoTDBInterpreterTest.java b/zeppelin-interpreter/src/test/java/org/apache/zeppelin/iotdb/IoTDBInterpreterTest.java
index 81f58a6..1525b3c 100644
--- a/zeppelin-interpreter/src/test/java/org/apache/zeppelin/iotdb/IoTDBInterpreterTest.java
+++ b/zeppelin-interpreter/src/test/java/org/apache/zeppelin/iotdb/IoTDBInterpreterTest.java
@@ -226,7 +226,7 @@ public class IoTDBInterpreterTest {
Assert.assertNotNull(actual);
Assert.assertEquals(Code.ERROR, actual.code());
Assert.assertEquals(
- "SQLException: 401: Error occurred while parsing SQL to physical plan: line 1:19 extraneous input 'a' expecting <EOF>",
+ "SQLException: 401: Error occurred while parsing SQL to physical plan: line 1:19 extraneous input 'a' expecting {<EOF>, ';'}",
actual.message().get(0).getData());
}