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 <[email protected]> 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()); }
