This is an automated email from the ASF dual-hosted git repository.
haonan pushed a commit to branch feature/continue_query
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/feature/continue_query by this
push:
new 9bfb98e add cq
new 1a2b4f9 Merge pull request #2988 from mzp0514/cq
9bfb98e is described below
commit 9bfb98e2f80360b613530872bf465eb6da2f878e
Author: mzp0514 <[email protected]>
AuthorDate: Tue Mar 30 21:00:23 2021 +0800
add cq
---
.../antlr4/org/apache/iotdb/db/qp/sql/SqlBase.g4 | 57 +++
.../iotdb/cluster/log/applier/BaseApplier.java | 6 +-
.../iotdb/cluster/log/applier/MetaLogApplier.java | 4 +-
.../main/java/org/apache/iotdb/SessionExample.java | 21 ++
.../org/apache/iotdb/db/auth/AuthorityChecker.java | 4 +
.../apache/iotdb/db/auth/entity/PrivilegeType.java | 2 +
.../org/apache/iotdb/db/conf/IoTDBConstant.java | 6 +
.../org/apache/iotdb/db/cq/ContinuousQuery.java | 292 +++++++++++++++
.../apache/iotdb/db/cq/ContinuousQueryService.java | 191 ++++++++++
.../ContinuousQueryAlreadyExistException.java} | 26 +-
.../ContinuousQueryNotExistException.java} | 26 +-
.../apache/iotdb/db/metadata/MLogTxtWriter.java | 25 ++
.../org/apache/iotdb/db/metadata/MManager.java | 39 +-
.../iotdb/db/metadata/MetadataOperationType.java | 2 +
.../iotdb/db/metadata/logfile/MLogWriter.java | 12 +
.../org/apache/iotdb/db/mqtt/PublishHandler.java | 4 +-
.../main/java/org/apache/iotdb/db/qp/Planner.java | 19 +
.../apache/iotdb/db/qp/constant/SQLConstant.java | 8 +
.../apache/iotdb/db/qp/executor/IPlanExecutor.java | 3 +-
.../apache/iotdb/db/qp/executor/PlanExecutor.java | 62 +++-
.../org/apache/iotdb/db/qp/logical/Operator.java | 5 +-
.../iotdb/db/qp/logical/crud/SelectOperator.java | 31 ++
.../logical/sys/CreateContinuousQueryOperator.java | 87 +++++
.../logical/sys/DropContinuousQueryOperator.java} | 30 +-
.../sys/ShowContinuousQueriesOperator.java} | 20 +-
.../apache/iotdb/db/qp/physical/PhysicalPlan.java | 14 +-
.../qp/physical/sys/CreateContinuousQueryPlan.java | 132 +++++++
.../qp/physical/sys/DropContinuousQueryPlan.java | 63 ++++
.../physical/sys/ShowContinuousQueriesPlan.java} | 19 +-
.../apache/iotdb/db/qp/physical/sys/ShowPlan.java | 3 +-
.../apache/iotdb/db/qp/sql/IoTDBSqlVisitor.java | 193 +++++++++-
.../iotdb/db/qp/strategy/PhysicalGenerator.java | 20 ++
.../query/dataset/ShowContinuousQueriesResult.java | 84 +++++
.../query/dataset/groupby/GroupByTimeDataSet.java | 11 +-
.../db/query/udf/core/context/UDFContext.java | 28 ++
.../java/org/apache/iotdb/db/service/IoTDB.java | 4 +
.../org/apache/iotdb/db/service/ServiceType.java | 3 +-
.../org/apache/iotdb/db/service/TSServiceImpl.java | 3 +-
.../org/apache/iotdb/db/tools/mlog/MLogParser.java | 15 +-
.../integration/IoTDBCreateContinuousQueryIT.java | 400 +++++++++++++++++++++
.../iotdb/db/qp/physical/PhysicalPlanTest.java | 124 +++++++
.../java/org/apache/iotdb/rpc/TSStatusCode.java | 2 +
42 files changed, 2003 insertions(+), 97 deletions(-)
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 0a3dbe3..9b43a9b 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
@@ -107,6 +107,33 @@ statement
fromClause
whereClause?
specialClause? #selectStatement
+ | CREATE CONTINUOUS QUERY continuousQueryName=ID
+ resampleClause?
+ BEGIN
+ cqSelectIntoClause
+ END # createContinuousQueryStatement
+ | DROP CONTINUOUS QUERY continuousQueryName=ID#
dropContinuousQueryStatement
+ | SHOW CONTINUOUS QUERIES # showContinuousQueriesStatement
+ ;
+
+
+resampleClause
+ : RESAMPLE (EVERY DURATION)? (FOR DURATION)?;
+
+cqSelectIntoClause
+ : SELECT selectElements
+ INTO (fullPath | suffixPath)
+ fromClause
+ whereClause?
+ cqGroupByTimeClause
+ ;
+
+
+cqGroupByTimeClause
+ : GROUP BY TIME LR_BRACKET
+ DURATION
+ RR_BRACKET
+ (COMMA LEVEL OPERATOR_EQ INT)?
;
selectElements
@@ -646,6 +673,11 @@ nodeNameWithoutStar
| PARTITION
| DESC
| ASC
+ | CONTINUOUS
+ | BEGIN
+ | END
+ | RESAMPLE
+ | EVERY
;
dataType
@@ -1290,6 +1322,31 @@ EXPLAIN
: E X P L A I N
;
+
+CONTINUOUS
+ : C O N T I N U O U S
+ ;
+
+QUERIES
+ : Q U E R I E S
+ ;
+
+BEGIN
+ : B E G I N
+ ;
+
+END
+ : E N D
+ ;
+
+RESAMPLE
+ : R E S A M P L E
+ ;
+
+EVERY
+ : E V E R Y
+ ;
+
//============================
// End of the keywords list
//============================
diff --git
a/cluster/src/main/java/org/apache/iotdb/cluster/log/applier/BaseApplier.java
b/cluster/src/main/java/org/apache/iotdb/cluster/log/applier/BaseApplier.java
index 460ba6e..c25d27e 100644
---
a/cluster/src/main/java/org/apache/iotdb/cluster/log/applier/BaseApplier.java
+++
b/cluster/src/main/java/org/apache/iotdb/cluster/log/applier/BaseApplier.java
@@ -65,7 +65,7 @@ abstract class BaseApplier implements LogApplier {
* @throws StorageEngineException
*/
void applyPhysicalPlan(PhysicalPlan plan, DataGroupMember dataGroupMember)
- throws QueryProcessException, StorageGroupNotSetException,
StorageEngineException {
+ throws QueryProcessException, MetadataException, StorageEngineException {
if (plan instanceof InsertPlan) {
processPlanWithTolerance((InsertPlan) plan, dataGroupMember);
} else if (plan != null && !plan.isQuery()) {
@@ -88,7 +88,7 @@ abstract class BaseApplier implements LogApplier {
}
private void executeAfterSync(PhysicalPlan plan)
- throws QueryProcessException, StorageGroupNotSetException,
StorageEngineException {
+ throws QueryProcessException, MetadataException, StorageEngineException {
try {
metaGroupMember.syncLeaderWithConsistencyCheck(true);
} catch (CheckConsistencyException ce) {
@@ -106,7 +106,7 @@ abstract class BaseApplier implements LogApplier {
* @throws StorageEngineException
*/
private void processPlanWithTolerance(InsertPlan plan, DataGroupMember
dataGroupMember)
- throws QueryProcessException, StorageGroupNotSetException,
StorageEngineException {
+ throws QueryProcessException, MetadataException, StorageEngineException {
try {
getQueryExecutor().processNonQuery(plan);
} catch (QueryProcessException | StorageGroupNotSetException |
StorageEngineException e) {
diff --git
a/cluster/src/main/java/org/apache/iotdb/cluster/log/applier/MetaLogApplier.java
b/cluster/src/main/java/org/apache/iotdb/cluster/log/applier/MetaLogApplier.java
index 0556ed7..6b907d5 100644
---
a/cluster/src/main/java/org/apache/iotdb/cluster/log/applier/MetaLogApplier.java
+++
b/cluster/src/main/java/org/apache/iotdb/cluster/log/applier/MetaLogApplier.java
@@ -26,7 +26,7 @@ import org.apache.iotdb.cluster.log.logtypes.RemoveNodeLog;
import org.apache.iotdb.cluster.rpc.thrift.Node;
import org.apache.iotdb.cluster.server.member.MetaGroupMember;
import org.apache.iotdb.db.exception.StorageEngineException;
-import org.apache.iotdb.db.exception.metadata.StorageGroupNotSetException;
+import org.apache.iotdb.db.exception.metadata.MetadataException;
import org.apache.iotdb.db.exception.query.QueryProcessException;
import org.slf4j.Logger;
@@ -59,7 +59,7 @@ public class MetaLogApplier extends BaseApplier {
} else {
logger.error("Unsupported log: {} {}", log.getClass().getName(), log);
}
- } catch (StorageEngineException | StorageGroupNotSetException |
QueryProcessException e) {
+ } catch (StorageEngineException | MetadataException |
QueryProcessException e) {
logger.debug("Exception occurred when executing {}", log, e);
log.setException(e);
} finally {
diff --git a/example/session/src/main/java/org/apache/iotdb/SessionExample.java
b/example/session/src/main/java/org/apache/iotdb/SessionExample.java
index 386b4e4..1116d47 100644
--- a/example/session/src/main/java/org/apache/iotdb/SessionExample.java
+++ b/example/session/src/main/java/org/apache/iotdb/SessionExample.java
@@ -71,6 +71,7 @@ public class SessionExample {
insertTablet();
insertTablets();
insertRecords();
+ createAndDropContinuousQueries();
nonQuery();
query();
queryWithTimeout();
@@ -93,6 +94,26 @@ public class SessionExample {
session.close();
}
+ private static void createAndDropContinuousQueries()
+ throws StatementExecutionException, IoTDBConnectionException {
+ session.executeNonQueryStatement(
+ "CREATE CONTINUOUS QUERY cq1 "
+ + "BEGIN SELECT max_value(s1) INTO temperature_max FROM root.sg1.*
"
+ + "GROUP BY time(10s) END");
+ session.executeNonQueryStatement(
+ "CREATE CONTINUOUS QUERY cq2 "
+ + "BEGIN SELECT count(s2) INTO temperature_cnt FROM root.sg1.* "
+ + "WHERE s2 > 80 GROUP BY time(10s), level=1 END");
+ session.executeNonQueryStatement(
+ "CREATE CONTINUOUS QUERY cq3 "
+ + "RESAMPLE EVERY 20s FOR 20s "
+ + "BEGIN SELECT avg(s3) INTO temperature_avg FROM root.sg1.* "
+ + "GROUP BY time(10s), level=1 END");
+ session.executeNonQueryStatement("DROP CONTINUOUS QUERY cq1");
+ session.executeNonQueryStatement("DROP CONTINUOUS QUERY cq2");
+ session.executeNonQueryStatement("DROP CONTINUOUS QUERY cq3");
+ }
+
private static void createTimeseries()
throws IoTDBConnectionException, StatementExecutionException {
diff --git
a/server/src/main/java/org/apache/iotdb/db/auth/AuthorityChecker.java
b/server/src/main/java/org/apache/iotdb/db/auth/AuthorityChecker.java
index fc5c680..b0dab97 100644
--- a/server/src/main/java/org/apache/iotdb/db/auth/AuthorityChecker.java
+++ b/server/src/main/java/org/apache/iotdb/db/auth/AuthorityChecker.java
@@ -157,6 +157,10 @@ public class AuthorityChecker {
return PrivilegeType.START_TRIGGER.ordinal();
case STOP_TRIGGER:
return PrivilegeType.STOP_TRIGGER.ordinal();
+ case CREATE_CONTINUOUS_QUERY:
+ return PrivilegeType.CREATE_CONTINUOUS_QUERY.ordinal();
+ case DROP_CONTINUOUS_QUERY:
+ return PrivilegeType.DROP_CONTINUOUS_QUERY.ordinal();
case AUTHOR:
case METADATA:
case BASIC_FUNC:
diff --git
a/server/src/main/java/org/apache/iotdb/db/auth/entity/PrivilegeType.java
b/server/src/main/java/org/apache/iotdb/db/auth/entity/PrivilegeType.java
index efde036..01dea3e 100644
--- a/server/src/main/java/org/apache/iotdb/db/auth/entity/PrivilegeType.java
+++ b/server/src/main/java/org/apache/iotdb/db/auth/entity/PrivilegeType.java
@@ -45,6 +45,8 @@ public enum PrivilegeType {
DROP_TRIGGER,
START_TRIGGER,
STOP_TRIGGER,
+ CREATE_CONTINUOUS_QUERY,
+ DROP_CONTINUOUS_QUERY,
ALL;
/**
diff --git a/server/src/main/java/org/apache/iotdb/db/conf/IoTDBConstant.java
b/server/src/main/java/org/apache/iotdb/db/conf/IoTDBConstant.java
index dd9422d..802833f 100644
--- a/server/src/main/java/org/apache/iotdb/db/conf/IoTDBConstant.java
+++ b/server/src/main/java/org/apache/iotdb/db/conf/IoTDBConstant.java
@@ -94,6 +94,12 @@ public class IoTDBConstant {
public static final String COLUMN_FUNCTION_TYPE = "function type";
public static final String COLUMN_FUNCTION_CLASS = "class name (UDF)";
+ public static final String COLUMN_CONTINUOUS_QUERY_NAME = "cq name";
+ public static final String COLUMN_CONTINUOUS_QUERY_EVERY_INTERVAL = "every
interval";
+ public static final String COLUMN_CONTINUOUS_QUERY_FOR_INTERVAL = "for
interval";
+ public static final String COLUMN_CONTINUOUS_QUERY_TARGET_PATH = "target
path";
+ public static final String COLUMN_CONTINUOUS_QUERY_QUERY_SQL = "query sql";
+
public static final String FUNCTION_TYPE_NATIVE = "native";
public static final String FUNCTION_TYPE_BUILTIN_UDAF = "built-in UDAF";
public static final String FUNCTION_TYPE_BUILTIN_UDTF = "built-in UDTF";
diff --git a/server/src/main/java/org/apache/iotdb/db/cq/ContinuousQuery.java
b/server/src/main/java/org/apache/iotdb/db/cq/ContinuousQuery.java
new file mode 100644
index 0000000..9ea3519
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/cq/ContinuousQuery.java
@@ -0,0 +1,292 @@
+/*
+ * 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.cq;
+
+import org.apache.iotdb.db.exception.StorageEngineException;
+import org.apache.iotdb.db.exception.metadata.IllegalPathException;
+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.Planner;
+import org.apache.iotdb.db.qp.constant.SQLConstant;
+import org.apache.iotdb.db.qp.executor.PlanExecutor;
+import org.apache.iotdb.db.qp.logical.crud.FilterOperator;
+import org.apache.iotdb.db.qp.logical.crud.QueryOperator;
+import org.apache.iotdb.db.qp.logical.crud.SelectOperator;
+import org.apache.iotdb.db.qp.physical.crud.GroupByTimePlan;
+import org.apache.iotdb.db.qp.physical.crud.InsertTabletPlan;
+import org.apache.iotdb.db.qp.physical.sys.CreateContinuousQueryPlan;
+import org.apache.iotdb.db.query.context.QueryContext;
+import org.apache.iotdb.db.query.control.QueryResourceManager;
+import
org.apache.iotdb.tsfile.exception.filter.QueryFilterOptimizationException;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.read.common.Field;
+import org.apache.iotdb.tsfile.read.common.Path;
+import org.apache.iotdb.tsfile.read.common.RowRecord;
+import org.apache.iotdb.tsfile.read.query.dataset.QueryDataSet;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+public class ContinuousQuery implements Runnable {
+
+ private final PlanExecutor planExecutor;
+ private final CreateContinuousQueryPlan plan;
+ private final Planner planner;
+
+ public ContinuousQuery(CreateContinuousQueryPlan plan) throws
QueryProcessException {
+ this.plan = plan;
+ this.planExecutor = new PlanExecutor();
+ this.planner = new Planner();
+ }
+
+ private static TSDataType getAggrDataType(String aggrFuncName, TSDataType
dataType) {
+ if (aggrFuncName == null) {
+ throw new IllegalArgumentException("AggregateFunction Name must not be
null");
+ }
+
+ switch (aggrFuncName.toLowerCase()) {
+ case SQLConstant.MIN_TIME:
+ case SQLConstant.MAX_TIME:
+ case SQLConstant.COUNT:
+ return TSDataType.INT64;
+ case SQLConstant.MIN_VALUE:
+ case SQLConstant.LAST_VALUE:
+ case SQLConstant.FIRST_VALUE:
+ case SQLConstant.MAX_VALUE:
+ return dataType;
+ case SQLConstant.AVG:
+ case SQLConstant.SUM:
+ return TSDataType.DOUBLE;
+ default:
+ throw new IllegalArgumentException("Invalid Aggregation function: " +
aggrFuncName);
+ }
+ }
+
+ @Override
+ public void run() {
+
+ try {
+
+ GroupByTimePlan queryPlan = getQueryPlan();
+
+ if (queryPlan.getDeduplicatedPaths().isEmpty()) {
+ return;
+ }
+
+ QueryDataSet result = doQuery(queryPlan);
+
+ if (result == null) {
+ return;
+ }
+
+ doInsert(result, queryPlan);
+
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+ }
+
+ private GroupByTimePlan getQueryPlan() throws QueryProcessException {
+ GroupByTimePlan queryPlan;
+
+ QueryOperator queryOperator = plan.getQueryOperator();
+
+ SelectOperator selectOperatorCopy =
queryOperator.getSelectOperator().copy();
+ FilterOperator filterOperatorCopy = null;
+ if (queryOperator.getFilterOperator() != null) {
+ filterOperatorCopy = queryOperator.getFilterOperator().copy();
+ }
+
+ queryPlan = (GroupByTimePlan)
planner.queryOperatorToPhysicalPlan(queryOperator, 1024);
+
+ queryOperator.setSelectOperator(selectOperatorCopy);
+ queryOperator.setFilterOperator(filterOperatorCopy);
+
+ long timestamp = System.currentTimeMillis();
+ queryPlan.setStartTime(timestamp - plan.getForInterval());
+ queryPlan.setEndTime(timestamp);
+
+ return queryPlan;
+ }
+
+ private QueryDataSet doQuery(GroupByTimePlan queryPlan)
+ throws StorageEngineException, QueryFilterOptimizationException,
MetadataException,
+ IOException, InterruptedException, QueryProcessException {
+ long queryId =
+ QueryResourceManager.getInstance()
+ .assignQueryId(true, 1024,
queryPlan.getDeduplicatedPaths().size());
+
+ QueryDataSet result;
+ result = planExecutor.processQuery(queryPlan, new QueryContext(queryId));
+ QueryResourceManager.getInstance().endQuery(queryId);
+ return result;
+ }
+
+ private void doInsert(QueryDataSet result, GroupByTimePlan queryPlan)
+ throws QueryProcessException, IOException, IllegalPathException {
+
+ int columnSize = result.getDataTypes().size();
+ TSDataType dataType =
+ getAggrDataType(queryPlan.getAggregations().get(0),
queryPlan.getDataTypes().get(0));
+
+ InsertTabletPlan[] insertTabletPlans = getInsertTabletPlans(columnSize,
result, dataType);
+
+ int fetchSize =
+ (int)
+ Math.min(
+ 10, Math.ceil((float) plan.getForInterval() /
plan.getQueryOperator().getUnit()));
+
+ Object[][] columns = getColumns(columnSize, fetchSize, dataType);
+ long[][] timestamps = new long[columnSize][fetchSize];
+ int[] rowNums = new int[columnSize];
+
+ boolean hasNext = true;
+
+ while (hasNext) {
+ int rowNum = 0;
+ Arrays.fill(rowNums, 0);
+
+ while (++rowNum <= fetchSize) {
+ if (!result.hasNextWithoutConstraint()) {
+ hasNext = false;
+ break;
+ }
+ RowRecord record = result.nextWithoutConstraint();
+ fillColumns(columns, dataType, record, rowNums, timestamps);
+ }
+
+ for (int i = 0; i < columnSize; i++) {
+ if (rowNums[i] > 0) {
+ insertTabletPlans[i].setTimes(timestamps[i]);
+ insertTabletPlans[i].setColumns(columns[i]);
+ insertTabletPlans[i].setRowCount(rowNums[i]);
+ planExecutor.insertTablet(insertTabletPlans[i]);
+ }
+ }
+ }
+ }
+
+ private InsertTabletPlan[] getInsertTabletPlans(
+ int columnSize, QueryDataSet result, TSDataType dataType) throws
IllegalPathException {
+ List<PartialPath> targetPaths = getTargetPaths(result.getPaths());
+ InsertTabletPlan[] insertTabletPlans = new InsertTabletPlan[columnSize];
+ String[] measurements = new String[] {targetPaths.get(0).getMeasurement()};
+ List<Integer> dataTypes = Collections.singletonList(dataType.ordinal());
+
+ for (int i = 0; i < columnSize; i++) {
+ insertTabletPlans[i] =
+ new InsertTabletPlan(
+ new PartialPath(targetPaths.get(i).getDevice()), measurements,
dataTypes);
+ }
+
+ return insertTabletPlans;
+ }
+
+ private Object[][] getColumns(int columnSize, int fetchSize, TSDataType
dataType) {
+ Object[][] columns = new Object[columnSize][1];
+ for (int i = 0; i < columnSize; i++) {
+ switch (dataType) {
+ case DOUBLE:
+ columns[i][0] = new double[fetchSize];
+ break;
+ case INT64:
+ columns[i][0] = new long[fetchSize];
+ break;
+ case INT32:
+ columns[i][0] = new int[fetchSize];
+ break;
+ case FLOAT:
+ columns[i][0] = new float[fetchSize];
+ break;
+ default:
+ break;
+ }
+ }
+ return columns;
+ }
+
+ private void fillColumns(
+ Object[][] columns,
+ TSDataType dataType,
+ RowRecord record,
+ int[] rowNums,
+ long[][] timestamps) {
+ List<Field> fields = record.getFields();
+ long ts = record.getTimestamp();
+
+ for (int i = 0; i < columns.length; i++) {
+ Field field = fields.get(i);
+ if (field != null) {
+ timestamps[i][rowNums[i]] = ts;
+ switch (dataType) {
+ case DOUBLE:
+ ((double[]) columns[i][0])[rowNums[i]] = field.getDoubleV();
+ break;
+ case INT64:
+ ((long[]) columns[i][0])[rowNums[i]] = field.getLongV();
+ break;
+ case INT32:
+ ((int[]) columns[i][0])[rowNums[i]] = field.getIntV();
+ break;
+ case FLOAT:
+ ((float[]) columns[i][0])[rowNums[i]] = field.getFloatV();
+ break;
+ default:
+ }
+
+ rowNums[i]++;
+ }
+ }
+ }
+
+ private List<PartialPath> getTargetPaths(List<Path> rawPaths) throws
IllegalPathException {
+ List<PartialPath> targetPaths = new ArrayList<>(rawPaths.size());
+ for (Path rawPath : rawPaths) {
+ targetPaths.add(new PartialPath(fillTemplate((PartialPath) rawPath)));
+ }
+ return targetPaths;
+ }
+
+ private String fillTemplate(PartialPath rawPath) {
+ String[] nodes = rawPath.getNodes();
+ int indexOfLeftBracket = nodes[0].indexOf("(");
+ if (indexOfLeftBracket != -1) {
+ nodes[0] = nodes[0].substring(indexOfLeftBracket + 1);
+ }
+ int indexOfRightBracket = nodes[nodes.length - 1].indexOf(")");
+ if (indexOfRightBracket != -1) {
+ nodes[nodes.length - 1] = nodes[nodes.length - 1].substring(0,
indexOfRightBracket);
+ }
+ StringBuffer sb = new StringBuffer();
+ Matcher m =
Pattern.compile("\\$\\{\\w+}").matcher(this.plan.getTargetPath().getFullPath());
+ while (m.find()) {
+ String param = m.group();
+ String value = nodes[Integer.parseInt(param.substring(2, param.length()
- 1).trim())];
+ m.appendReplacement(sb, value == null ? "" : value);
+ }
+ m.appendTail(sb);
+ return sb.toString();
+ }
+}
diff --git
a/server/src/main/java/org/apache/iotdb/db/cq/ContinuousQueryService.java
b/server/src/main/java/org/apache/iotdb/db/cq/ContinuousQueryService.java
new file mode 100644
index 0000000..0f4ec15
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/cq/ContinuousQueryService.java
@@ -0,0 +1,191 @@
+/*
+ * 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.cq;
+
+import org.apache.iotdb.db.concurrent.IoTDBThreadPoolFactory;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import
org.apache.iotdb.db.exception.metadata.ContinuousQueryAlreadyExistException;
+import org.apache.iotdb.db.exception.metadata.ContinuousQueryNotExistException;
+import org.apache.iotdb.db.exception.query.QueryProcessException;
+import org.apache.iotdb.db.qp.physical.sys.CreateContinuousQueryPlan;
+import org.apache.iotdb.db.qp.physical.sys.DropContinuousQueryPlan;
+import org.apache.iotdb.db.query.dataset.ShowContinuousQueriesResult;
+import org.apache.iotdb.db.service.IService;
+import org.apache.iotdb.db.service.IoTDB;
+import org.apache.iotdb.db.service.ServiceType;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.ReentrantLock;
+
+public class ContinuousQueryService implements IService {
+
+ private static final Logger logger =
LoggerFactory.getLogger(ContinuousQueryService.class);
+
+ private final HashMap<String, ScheduledFuture<?>> continuousQueriesFutures =
new HashMap<>();
+
+ private final HashMap<String, CreateContinuousQueryPlan>
continuousQueryPlans = new HashMap<>();
+
+ private final ScheduledExecutorService pool =
+ IoTDBThreadPoolFactory.newScheduledThreadPool(10, "Continuous Query
Service");
+
+ private final ReentrantLock registrationLock = new ReentrantLock();
+
+ private static final ContinuousQueryService INSTANCE = new
ContinuousQueryService();
+
+ public static ContinuousQueryService getInstance() {
+ return INSTANCE;
+ }
+
+ public void acquireRegistrationLock() {
+ registrationLock.lock();
+ }
+
+ public void releaseRegistrationLock() {
+ registrationLock.unlock();
+ }
+
+ @Override
+ public ServiceType getID() {
+ return ServiceType.CONTINUOUS_QUERY_SERVICE;
+ }
+
+ @Override
+ public void start() {
+ logger.info("Continuous query service started.");
+ }
+
+ @Override
+ public void stop() {
+ pool.shutdownNow();
+ }
+
+ @Override
+ public void waitAndStop(long milliseconds) {
+ for (ScheduledFuture<?> future : continuousQueriesFutures.values()) {
+ future.cancel(false);
+ }
+ logger.info("Waiting for task pool to shut down");
+ try {
+ Thread.sleep(milliseconds);
+ } catch (InterruptedException e) {
+ logger.info("Thread interrupted");
+ Thread.currentThread().interrupt();
+ }
+ pool.shutdownNow();
+ }
+
+ public boolean register(CreateContinuousQueryPlan plan, boolean writeLog)
+ throws ContinuousQueryAlreadyExistException {
+
+ if (continuousQueryPlans.containsKey(plan.getContinuousQueryName())) {
+ throw new
ContinuousQueryAlreadyExistException(plan.getContinuousQueryName());
+ }
+
+ if (writeLog) {
+ try {
+ IoTDB.metaManager.createContinuousQuery(plan);
+ } catch (IOException e) {
+ e.printStackTrace();
+ }
+ }
+
+ doRegister(plan);
+ return true;
+ }
+
+ private void doRegister(CreateContinuousQueryPlan plan) {
+ acquireRegistrationLock();
+
+ try {
+
+ ContinuousQuery cq = new ContinuousQuery(plan);
+ ScheduledFuture<?> future =
+ pool.scheduleAtFixedRate(
+ cq, plan.getEveryInterval(), plan.getEveryInterval(),
getTimeUnit());
+ continuousQueriesFutures.put(plan.getContinuousQueryName(), future);
+ continuousQueryPlans.put(plan.getContinuousQueryName(), plan);
+ } catch (QueryProcessException e) {
+ e.printStackTrace();
+ } finally {
+ releaseRegistrationLock();
+ }
+ }
+
+ private TimeUnit getTimeUnit() {
+ String timestampPrecision =
IoTDBDescriptor.getInstance().getConfig().getTimestampPrecision();
+ if (timestampPrecision.equals("us")) {
+ return TimeUnit.MICROSECONDS;
+ } else if (timestampPrecision.equals("ns")) {
+ return TimeUnit.NANOSECONDS;
+ } else {
+ return TimeUnit.MILLISECONDS;
+ }
+ }
+
+ public boolean deregister(DropContinuousQueryPlan plan) throws
ContinuousQueryNotExistException {
+ if (!continuousQueryPlans.containsKey(plan.getContinuousQueryName())) {
+ throw new
ContinuousQueryNotExistException(plan.getContinuousQueryName());
+ }
+
+ try {
+
+ IoTDB.metaManager.dropContinuousQuery(plan);
+ doDeregister(plan);
+ } catch (Exception e) {
+
+ e.printStackTrace();
+ }
+
+ return true;
+ }
+
+ private void doDeregister(DropContinuousQueryPlan plan) {
+ String cqName = plan.getContinuousQueryName();
+ continuousQueriesFutures.get(cqName).cancel(false);
+ continuousQueriesFutures.remove(cqName);
+ continuousQueryPlans.remove(cqName);
+ }
+
+ public List<ShowContinuousQueriesResult> getContinuousQueryPlans() {
+
+ List<ShowContinuousQueriesResult> results = new
ArrayList<>(continuousQueryPlans.size());
+
+ for (CreateContinuousQueryPlan plan : continuousQueryPlans.values()) {
+ results.add(
+ new ShowContinuousQueriesResult(
+ plan.getQuerySql(),
+ plan.getContinuousQueryName(),
+ plan.getTargetPath(),
+ plan.getEveryInterval(),
+ plan.getForInterval()));
+ }
+
+ return results;
+ }
+}
diff --git
a/server/src/main/java/org/apache/iotdb/db/metadata/MetadataOperationType.java
b/server/src/main/java/org/apache/iotdb/db/exception/metadata/ContinuousQueryAlreadyExistException.java
similarity index 57%
copy from
server/src/main/java/org/apache/iotdb/db/metadata/MetadataOperationType.java
copy to
server/src/main/java/org/apache/iotdb/db/exception/metadata/ContinuousQueryAlreadyExistException.java
index 466c0a2..3f07589 100644
---
a/server/src/main/java/org/apache/iotdb/db/metadata/MetadataOperationType.java
+++
b/server/src/main/java/org/apache/iotdb/db/exception/metadata/ContinuousQueryAlreadyExistException.java
@@ -16,21 +16,19 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.iotdb.db.metadata;
-public class MetadataOperationType {
+package org.apache.iotdb.db.exception.metadata;
- private MetadataOperationType() {
- // allowed to do nothing
- }
+import org.apache.iotdb.rpc.TSStatusCode;
+
+public class ContinuousQueryAlreadyExistException extends MetadataException {
- public static final String CREATE_TIMESERIES = "0";
- public static final String DELETE_TIMESERIES = "1";
- public static final String SET_STORAGE_GROUP = "2";
- public static final String SET_TTL = "10";
- public static final String DELETE_STORAGE_GROUP = "11";
- public static final String CREATE_INDEX = "31";
- public static final String DROP_INDEX = "32";
- public static final String CHANGE_OFFSET = "12";
- public static final String CHANGE_ALIAS = "13";
+ private static final long serialVersionUID = -6713847897890531438L;
+
+ public ContinuousQueryAlreadyExistException(String continuousQueryName) {
+ super(
+ String.format("Continuous Query [%s] already exist",
continuousQueryName),
+ TSStatusCode.PATH_ALREADY_EXIST_ERROR.getStatusCode());
+ this.isUserException = true;
+ }
}
diff --git
a/server/src/main/java/org/apache/iotdb/db/metadata/MetadataOperationType.java
b/server/src/main/java/org/apache/iotdb/db/exception/metadata/ContinuousQueryNotExistException.java
similarity index 57%
copy from
server/src/main/java/org/apache/iotdb/db/metadata/MetadataOperationType.java
copy to
server/src/main/java/org/apache/iotdb/db/exception/metadata/ContinuousQueryNotExistException.java
index 466c0a2..f99685c 100644
---
a/server/src/main/java/org/apache/iotdb/db/metadata/MetadataOperationType.java
+++
b/server/src/main/java/org/apache/iotdb/db/exception/metadata/ContinuousQueryNotExistException.java
@@ -16,21 +16,19 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.iotdb.db.metadata;
-public class MetadataOperationType {
+package org.apache.iotdb.db.exception.metadata;
- private MetadataOperationType() {
- // allowed to do nothing
- }
+import org.apache.iotdb.rpc.TSStatusCode;
+
+public class ContinuousQueryNotExistException extends MetadataException {
- public static final String CREATE_TIMESERIES = "0";
- public static final String DELETE_TIMESERIES = "1";
- public static final String SET_STORAGE_GROUP = "2";
- public static final String SET_TTL = "10";
- public static final String DELETE_STORAGE_GROUP = "11";
- public static final String CREATE_INDEX = "31";
- public static final String DROP_INDEX = "32";
- public static final String CHANGE_OFFSET = "12";
- public static final String CHANGE_ALIAS = "13";
+ private static final long serialVersionUID = -6713847897890531438L;
+
+ public ContinuousQueryNotExistException(String continuousQueryName) {
+ super(
+ String.format("Continuous Query [%s] does not exist",
continuousQueryName),
+ TSStatusCode.CONTINUOUS_QUERY_NOT_EXIST.getStatusCode());
+ this.isUserException = true;
+ }
}
diff --git
a/server/src/main/java/org/apache/iotdb/db/metadata/MLogTxtWriter.java
b/server/src/main/java/org/apache/iotdb/db/metadata/MLogTxtWriter.java
index ebfa2c6..a14ffb5 100644
--- a/server/src/main/java/org/apache/iotdb/db/metadata/MLogTxtWriter.java
+++ b/server/src/main/java/org/apache/iotdb/db/metadata/MLogTxtWriter.java
@@ -19,6 +19,8 @@
package org.apache.iotdb.db.metadata;
import org.apache.iotdb.db.engine.fileSystem.SystemFileFactory;
+import org.apache.iotdb.db.qp.physical.sys.CreateContinuousQueryPlan;
+import org.apache.iotdb.db.qp.physical.sys.DropContinuousQueryPlan;
import org.apache.iotdb.db.qp.physical.sys.CreateTimeSeriesPlan;
import org.apache.iotdb.db.qp.physical.sys.MNodePlan;
import org.apache.iotdb.db.qp.physical.sys.MeasurementMNodePlan;
@@ -126,6 +128,29 @@ public class MLogTxtWriter implements AutoCloseable {
channel.write(buff);
}
+ public void createContinuousQuery(CreateContinuousQueryPlan plan) throws
IOException {
+ String buf =
+ String.format(
+ "%s,%s,%s,%s",
+ MetadataOperationType.CREATE_CONTINUOUS_QUERY,
+ plan.getContinuousQueryName(),
+ plan.getQuerySql(),
+ plan.getTargetPath().getFullPath())
+ + LINE_SEPARATOR;
+ channel.write(ByteBuffer.wrap(buf.getBytes()));
+ lineNumber.incrementAndGet();
+ }
+
+ public void dropContinuousQuery(DropContinuousQueryPlan plan) throws
IOException {
+
+ String buf =
+ String.format(
+ "%s,%s", MetadataOperationType.DROP_CONTINUOUS_QUERY,
plan.getContinuousQueryName())
+ + LINE_SEPARATOR;
+ channel.write(ByteBuffer.wrap(buf.getBytes()));
+ lineNumber.incrementAndGet();
+ }
+
public void setStorageGroup(String storageGroup) throws IOException {
String outputStr =
MetadataOperationType.SET_STORAGE_GROUP + "," + storageGroup +
LINE_SEPARATOR;
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 c6bcb78..3ed8241 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
@@ -20,6 +20,7 @@ package org.apache.iotdb.db.metadata;
import org.apache.iotdb.db.conf.IoTDBConfig;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.cq.ContinuousQueryService;
import org.apache.iotdb.db.engine.StorageEngine;
import org.apache.iotdb.db.engine.fileSystem.SystemFileFactory;
import org.apache.iotdb.db.engine.storagegroup.StorageGroupProcessor;
@@ -40,10 +41,14 @@ import org.apache.iotdb.db.metadata.mnode.MeasurementMNode;
import org.apache.iotdb.db.metadata.mnode.StorageGroupMNode;
import org.apache.iotdb.db.monitor.MonitorConstants;
import org.apache.iotdb.db.qp.constant.SQLConstant;
+import org.apache.iotdb.db.qp.logical.Operator;
+import org.apache.iotdb.db.qp.logical.crud.QueryOperator;
import org.apache.iotdb.db.qp.physical.PhysicalPlan;
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.sys.CreateContinuousQueryPlan;
+import org.apache.iotdb.db.qp.physical.sys.DropContinuousQueryPlan;
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.CreateTimeSeriesPlan;
@@ -53,6 +58,7 @@ import
org.apache.iotdb.db.qp.physical.sys.SetStorageGroupPlan;
import org.apache.iotdb.db.qp.physical.sys.SetTTLPlan;
import org.apache.iotdb.db.qp.physical.sys.ShowDevicesPlan;
import org.apache.iotdb.db.qp.physical.sys.ShowTimeSeriesPlan;
+import org.apache.iotdb.db.qp.strategy.LogicalGenerator;
import org.apache.iotdb.db.query.context.QueryContext;
import org.apache.iotdb.db.query.dataset.ShowDevicesResult;
import org.apache.iotdb.db.query.dataset.ShowTimeSeriesResult;
@@ -78,6 +84,7 @@ import org.slf4j.LoggerFactory;
import java.io.File;
import java.io.IOException;
import java.nio.file.Files;
+import java.time.ZoneId;
import java.util.ArrayDeque;
import java.util.ArrayList;
import java.util.Collection;
@@ -287,6 +294,7 @@ public class MManager {
private int applyMlog(MLogReader mLogReader) {
int idx = 0;
+ HashMap<String, CreateContinuousQueryPlan> recoveredCQs = new HashMap<>();
while (mLogReader.hasNext()) {
PhysicalPlan plan = null;
try {
@@ -294,13 +302,34 @@ public class MManager {
if (plan == null) {
continue;
}
- operation(plan);
+ if (plan.getOperatorType() ==
Operator.OperatorType.CREATE_CONTINUOUS_QUERY) {
+ recoveredCQs.put(
+ ((CreateContinuousQueryPlan) plan).getContinuousQueryName(),
+ (CreateContinuousQueryPlan) plan);
+ } else if (plan.getOperatorType() ==
Operator.OperatorType.DROP_CONTINUOUS_QUERY) {
+ recoveredCQs.remove(((DropContinuousQueryPlan)
plan).getContinuousQueryName());
+ } else {
+ operation(plan);
+ }
idx++;
} catch (Exception e) {
logger.error(
"Can not operate cmd {} for err:", plan == null ? "" :
plan.getOperatorType(), e);
}
}
+
+ LogicalGenerator logicalGenerator = new LogicalGenerator();
+ for (Map.Entry<String, CreateContinuousQueryPlan> cq :
recoveredCQs.entrySet()) {
+ CreateContinuousQueryPlan plan = cq.getValue();
+ QueryOperator queryOperator =
+ (QueryOperator) logicalGenerator.generate(plan.getQuerySql(),
ZoneId.systemDefault());
+ plan.setQueryOperator(queryOperator);
+ try {
+ ContinuousQueryService.getInstance().register(plan, false);
+ } catch (MetadataException e) {
+ e.printStackTrace();
+ }
+ }
return idx;
}
@@ -371,6 +400,14 @@ public class MManager {
}
}
+ public void createContinuousQuery(CreateContinuousQueryPlan plan) throws
IOException {
+ logWriter.createContinuousQuery(plan);
+ }
+
+ public void dropContinuousQuery(DropContinuousQueryPlan plan) throws
IOException {
+ logWriter.dropContinuousQuery(plan);
+ }
+
public void createTimeseries(CreateTimeSeriesPlan plan) throws
MetadataException {
createTimeseries(plan, -1);
}
diff --git
a/server/src/main/java/org/apache/iotdb/db/metadata/MetadataOperationType.java
b/server/src/main/java/org/apache/iotdb/db/metadata/MetadataOperationType.java
index 466c0a2..fe25640 100644
---
a/server/src/main/java/org/apache/iotdb/db/metadata/MetadataOperationType.java
+++
b/server/src/main/java/org/apache/iotdb/db/metadata/MetadataOperationType.java
@@ -33,4 +33,6 @@ public class MetadataOperationType {
public static final String DROP_INDEX = "32";
public static final String CHANGE_OFFSET = "12";
public static final String CHANGE_ALIAS = "13";
+ public static final String CREATE_CONTINUOUS_QUERY = "14";
+ public static final String DROP_CONTINUOUS_QUERY = "15";
}
diff --git
a/server/src/main/java/org/apache/iotdb/db/metadata/logfile/MLogWriter.java
b/server/src/main/java/org/apache/iotdb/db/metadata/logfile/MLogWriter.java
index 8fda60d..cddda60 100644
--- a/server/src/main/java/org/apache/iotdb/db/metadata/logfile/MLogWriter.java
+++ b/server/src/main/java/org/apache/iotdb/db/metadata/logfile/MLogWriter.java
@@ -28,6 +28,8 @@ 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.qp.physical.PhysicalPlan;
+import org.apache.iotdb.db.qp.physical.sys.CreateContinuousQueryPlan;
+import org.apache.iotdb.db.qp.physical.sys.DropContinuousQueryPlan;
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.CreateTimeSeriesPlan;
@@ -125,6 +127,16 @@ public class MLogWriter implements AutoCloseable {
putLog(deleteTimeSeriesPlan);
}
+ public void createContinuousQuery(CreateContinuousQueryPlan
createContinuousQueryPlan)
+ throws IOException {
+ putLog(createContinuousQueryPlan);
+ }
+
+ public void dropContinuousQuery(DropContinuousQueryPlan
dropContinuousQueryPlan)
+ throws IOException {
+ putLog(dropContinuousQueryPlan);
+ }
+
public void setStorageGroup(PartialPath storageGroup) throws IOException {
SetStorageGroupPlan plan = new SetStorageGroupPlan(storageGroup);
putLog(plan);
diff --git a/server/src/main/java/org/apache/iotdb/db/mqtt/PublishHandler.java
b/server/src/main/java/org/apache/iotdb/db/mqtt/PublishHandler.java
index 1e4b213..27e1579 100644
--- a/server/src/main/java/org/apache/iotdb/db/mqtt/PublishHandler.java
+++ b/server/src/main/java/org/apache/iotdb/db/mqtt/PublishHandler.java
@@ -20,7 +20,7 @@ package org.apache.iotdb.db.mqtt;
import org.apache.iotdb.db.conf.IoTDBConfig;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.exception.StorageEngineException;
-import org.apache.iotdb.db.exception.metadata.StorageGroupNotSetException;
+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.executor.IPlanExecutor;
@@ -118,7 +118,7 @@ public class PublishHandler extends
AbstractInterceptHandler {
}
private boolean executeNonQuery(PhysicalPlan plan)
- throws QueryProcessException, StorageGroupNotSetException,
StorageEngineException {
+ throws QueryProcessException, MetadataException, StorageEngineException {
if (IoTDBDescriptor.getInstance().getConfig().isReadOnly()) {
throw new QueryProcessException(
"Current system mode is read-only, does not support non-query
operation");
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/Planner.java
b/server/src/main/java/org/apache/iotdb/db/qp/Planner.java
index 355b7ef..df8c3ec 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/Planner.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/Planner.java
@@ -82,6 +82,22 @@ public class Planner {
return physicalGenerator.transformToPhysicalPlan(operator, fetchSize);
}
+ public PhysicalPlan queryOperatorToPhysicalPlan(SFWOperator queryOperator,
int fetchSize)
+ throws QueryProcessException {
+ int maxDeduplicatedPathNum =
+
QueryResourceManager.getInstance().getMaxDeduplicatedPathNum(fetchSize);
+ if (queryOperator.isLastQuery()) {
+ // Dataset of last query actually has only three columns, so we
shouldn't limit the path num
+ // while constructing logical plan
+ // To avoid overflowing because logicalOptimize function may do
maxDeduplicatedPathNum + 1, we
+ // set it to Integer.MAX_VALUE - 1
+ maxDeduplicatedPathNum = Integer.MAX_VALUE - 1;
+ }
+ queryOperator = optimizeSFWOperator(queryOperator, maxDeduplicatedPathNum);
+ PhysicalGenerator physicalGenerator = new PhysicalGenerator();
+ return physicalGenerator.transformToPhysicalPlan(queryOperator, fetchSize);
+ }
+
/** convert raw data query to physical plan directly */
public PhysicalPlan rawDataQueryReqToPhysicalPlan(
TSRawDataQueryReq rawDataQueryReq, ZoneId zoneId)
@@ -181,6 +197,9 @@ public class Planner {
case DROP_TRIGGER:
case START_TRIGGER:
case STOP_TRIGGER:
+ case CREATE_CONTINUOUS_QUERY:
+ case SHOW_CONTINUOUS_QUERIES:
+ case DROP_CONTINUOUS_QUERY:
return operator;
case QUERY:
case DELETE:
diff --git
a/server/src/main/java/org/apache/iotdb/db/qp/constant/SQLConstant.java
b/server/src/main/java/org/apache/iotdb/db/qp/constant/SQLConstant.java
index 35dcdec..8ee51c7 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/constant/SQLConstant.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/constant/SQLConstant.java
@@ -178,6 +178,10 @@ public class SQLConstant {
public static final int TOK_TRIGGER_STOP = 103;
public static final int TOK_SHOW_TRIGGERS = 104;
+ public static final int TOK_CONTINUOUS_QUERY_CREATE = 105;
+ public static final int TOK_CONTINUOUS_QUERY_DROP = 106;
+ public static final int TOK_SHOW_CONTINUOUS_QUERIES = 107;
+
public static final Map<Integer, String> tokenSymbol = new HashMap<>();
public static final Map<Integer, String> tokenNames = new HashMap<>();
public static final Map<Integer, Integer> reverseWords = new HashMap<>();
@@ -271,6 +275,10 @@ public class SQLConstant {
tokenNames.put(TOK_TRIGGER_START, "TOK_TRIGGER_START");
tokenNames.put(TOK_TRIGGER_STOP, "TOK_TRIGGER_STOP");
tokenNames.put(TOK_SHOW_TRIGGERS, "TOK_SHOW_TRIGGERS");
+
+ tokenNames.put(TOK_CONTINUOUS_QUERY_CREATE, "TOK_CONTINUOUS_QUERY_CREATE");
+ tokenNames.put(TOK_CONTINUOUS_QUERY_DROP, "TOK_CONTINUOUS_QUERY_DROP");
+ tokenNames.put(TOK_SHOW_CONTINUOUS_QUERIES, "TOK_SHOW_CONTINUOUS_QUERIES");
}
static {
diff --git
a/server/src/main/java/org/apache/iotdb/db/qp/executor/IPlanExecutor.java
b/server/src/main/java/org/apache/iotdb/db/qp/executor/IPlanExecutor.java
index af24983..f302bdb 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/executor/IPlanExecutor.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/executor/IPlanExecutor.java
@@ -59,7 +59,8 @@ public interface IPlanExecutor {
* @param plan Physical Non-Query Plan
*/
boolean processNonQuery(PhysicalPlan plan)
- throws QueryProcessException, StorageGroupNotSetException,
StorageEngineException;
+ throws QueryProcessException, StorageGroupNotSetException,
StorageEngineException,
+ MetadataException;
/**
* execute update command and return whether the operator is successful.
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 54a0bf7..221b5a8 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
@@ -27,6 +27,7 @@ import org.apache.iotdb.db.auth.entity.Role;
import org.apache.iotdb.db.auth.entity.User;
import org.apache.iotdb.db.conf.IoTDBConstant;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.cq.ContinuousQueryService;
import org.apache.iotdb.db.engine.StorageEngine;
import org.apache.iotdb.db.engine.cache.ChunkCache;
import org.apache.iotdb.db.engine.cache.TimeSeriesMetadataCache;
@@ -43,6 +44,8 @@ import
org.apache.iotdb.db.exception.metadata.IllegalPathException;
import org.apache.iotdb.db.exception.metadata.MetadataException;
import org.apache.iotdb.db.exception.metadata.PathNotExistException;
import org.apache.iotdb.db.exception.metadata.StorageGroupNotSetException;
+import
org.apache.iotdb.db.exception.metadata.ContinuousQueryAlreadyExistException;
+import org.apache.iotdb.db.exception.metadata.ContinuousQueryNotExistException;
import org.apache.iotdb.db.exception.query.QueryProcessException;
import org.apache.iotdb.db.metadata.PartialPath;
import org.apache.iotdb.db.metadata.mnode.MNode;
@@ -72,6 +75,9 @@ 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.UDTFPlan;
+import org.apache.iotdb.db.qp.physical.sys.CreateContinuousQueryPlan;
+import org.apache.iotdb.db.qp.physical.sys.DropContinuousQueryPlan;
+import org.apache.iotdb.db.qp.physical.sys.ShowContinuousQueriesPlan;
import org.apache.iotdb.db.qp.physical.sys.AlterTimeSeriesPlan;
import org.apache.iotdb.db.qp.physical.sys.AuthorPlan;
import org.apache.iotdb.db.qp.physical.sys.CountPlan;
@@ -111,6 +117,7 @@ import org.apache.iotdb.db.query.dataset.ListDataSet;
import org.apache.iotdb.db.query.dataset.ShowDevicesDataSet;
import org.apache.iotdb.db.query.dataset.ShowTimeseriesDataSet;
import org.apache.iotdb.db.query.dataset.SingleDataSet;
+import org.apache.iotdb.db.query.dataset.ShowContinuousQueriesResult;
import org.apache.iotdb.db.query.executor.IQueryRouter;
import org.apache.iotdb.db.query.executor.QueryRouter;
import org.apache.iotdb.db.query.udf.service.UDFRegistrationInformation;
@@ -185,6 +192,11 @@ import static
org.apache.iotdb.db.conf.IoTDBConstant.FUNCTION_TYPE_NATIVE;
import static org.apache.iotdb.db.conf.IoTDBConstant.QUERY_ID;
import static org.apache.iotdb.db.conf.IoTDBConstant.STATEMENT;
import static
org.apache.iotdb.tsfile.common.constant.TsFileConstant.TSFILE_SUFFIX;
+import static
org.apache.iotdb.db.conf.IoTDBConstant.COLUMN_CONTINUOUS_QUERY_EVERY_INTERVAL;
+import static
org.apache.iotdb.db.conf.IoTDBConstant.COLUMN_CONTINUOUS_QUERY_FOR_INTERVAL;
+import static
org.apache.iotdb.db.conf.IoTDBConstant.COLUMN_CONTINUOUS_QUERY_NAME;
+import static
org.apache.iotdb.db.conf.IoTDBConstant.COLUMN_CONTINUOUS_QUERY_QUERY_SQL;
+import static
org.apache.iotdb.db.conf.IoTDBConstant.COLUMN_CONTINUOUS_QUERY_TARGET_PATH;
@SuppressWarnings("java:S1135") // ignore todos
public class PlanExecutor implements IPlanExecutor {
@@ -222,7 +234,7 @@ public class PlanExecutor implements IPlanExecutor {
@Override
public boolean processNonQuery(PhysicalPlan plan)
- throws QueryProcessException, StorageGroupNotSetException,
StorageEngineException {
+ throws QueryProcessException, StorageEngineException, MetadataException {
switch (plan.getOperatorType()) {
case DELETE:
delete((DeletePlan) plan);
@@ -335,6 +347,10 @@ public class PlanExecutor implements IPlanExecutor {
throw new QueryProcessException(e.getMessage());
}
return true;
+ case CREATE_CONTINUOUS_QUERY:
+ return operateCreateContinuousQuery((CreateContinuousQueryPlan) plan);
+ case DROP_CONTINUOUS_QUERY:
+ return operateDropContinuousQuery((DropContinuousQueryPlan) plan);
default:
throw new UnsupportedOperationException(
String.format("operation %s is not supported",
plan.getOperatorType()));
@@ -430,6 +446,16 @@ public class PlanExecutor implements IPlanExecutor {
}
}
+ private boolean operateCreateContinuousQuery(CreateContinuousQueryPlan plan)
+ throws ContinuousQueryAlreadyExistException {
+ return ContinuousQueryService.getInstance().register(plan, true);
+ }
+
+ private boolean operateDropContinuousQuery(DropContinuousQueryPlan plan)
+ throws ContinuousQueryNotExistException {
+ return ContinuousQueryService.getInstance().deregister(plan);
+ }
+
public static void flushSpecifiedStorageGroups(FlushPlan plan)
throws StorageGroupNotSetException {
Map<PartialPath, List<Pair<Long, Boolean>>> storageGroupMap =
@@ -539,6 +565,8 @@ public class PlanExecutor implements IPlanExecutor {
return processShowFunctions((ShowFunctionsPlan) showPlan);
case TRIGGERS:
return processShowTriggers((ShowTriggersPlan) showPlan);
+ case CONTINUOUS_QUERY:
+ return processShowContinuousQueries();
default:
throw new QueryProcessException(String.format("Unrecognized show plan
%s", showPlan));
}
@@ -859,6 +887,38 @@ public class PlanExecutor implements IPlanExecutor {
}
}
+ private QueryDataSet processShowContinuousQueries() {
+ ListDataSet listDataSet =
+ new ListDataSet(
+ Arrays.asList(
+ new PartialPath(COLUMN_CONTINUOUS_QUERY_NAME, false),
+ new PartialPath(COLUMN_CONTINUOUS_QUERY_EVERY_INTERVAL, false),
+ new PartialPath(COLUMN_CONTINUOUS_QUERY_FOR_INTERVAL, false),
+ new PartialPath(COLUMN_CONTINUOUS_QUERY_QUERY_SQL, false),
+ new PartialPath(COLUMN_CONTINUOUS_QUERY_TARGET_PATH, false)),
+ Arrays.asList(
+ TSDataType.TEXT,
+ TSDataType.INT64,
+ TSDataType.INT64,
+ TSDataType.TEXT,
+ TSDataType.TEXT));
+
+ List<ShowContinuousQueriesResult> continuousQueriesList =
+ ContinuousQueryService.getInstance().getContinuousQueryPlans();
+
+ for (ShowContinuousQueriesResult result : continuousQueriesList) {
+ RowRecord record = new RowRecord(0);
+ record.addField(Binary.valueOf(result.getContinuousQueryName()),
TSDataType.TEXT);
+ record.addField(result.getEveryInterval(), TSDataType.INT64);
+ record.addField(result.getForInterval(), TSDataType.INT64);
+ record.addField(Binary.valueOf(result.getQuerySql()), TSDataType.TEXT);
+ record.addField(Binary.valueOf(result.getTargetPath().getFullPath()),
TSDataType.TEXT);
+ listDataSet.putRecord(record);
+ }
+
+ return listDataSet;
+ }
+
private void appendNativeFunctions(ListDataSet listDataSet,
ShowFunctionsPlan showPlan) {
if (showPlan.showTemporary()) {
return;
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 f2a8633..6a0514f 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
@@ -161,6 +161,9 @@ public abstract class Operator {
CREATE_TRIGGER,
DROP_TRIGGER,
START_TRIGGER,
- STOP_TRIGGER
+ STOP_TRIGGER,
+ CREATE_CONTINUOUS_QUERY,
+ DROP_CONTINUOUS_QUERY,
+ SHOW_CONTINUOUS_QUERIES
}
}
diff --git
a/server/src/main/java/org/apache/iotdb/db/qp/logical/crud/SelectOperator.java
b/server/src/main/java/org/apache/iotdb/db/qp/logical/crud/SelectOperator.java
index 5b34ea5..5790b80 100644
---
a/server/src/main/java/org/apache/iotdb/db/qp/logical/crud/SelectOperator.java
+++
b/server/src/main/java/org/apache/iotdb/db/qp/logical/crud/SelectOperator.java
@@ -18,6 +18,7 @@
*/
package org.apache.iotdb.db.qp.logical.crud;
+import org.apache.iotdb.db.exception.metadata.IllegalPathException;
import org.apache.iotdb.db.metadata.PartialPath;
import org.apache.iotdb.db.qp.logical.Operator;
import org.apache.iotdb.db.query.udf.core.context.UDFContext;
@@ -113,4 +114,34 @@ public final class SelectOperator extends Operator {
public void setUdfList(List<UDFContext> udfList) {
this.udfList = udfList;
}
+
+ public SelectOperator copy() {
+ SelectOperator ret = new SelectOperator(this.tokenIntType, this.zoneId);
+
+ try {
+ for (PartialPath path : this.suffixList) {
+ ret.suffixList.add(new PartialPath(path.getFullPath()));
+ }
+ } catch (IllegalPathException e) {
+ e.printStackTrace();
+ }
+
+ for (String aggr : this.aggregations) {
+ ret.aggregations.add(aggr);
+ }
+
+ for (UDFContext ctx : this.udfList) {
+ if (ctx != null) {
+ ret.udfList.add(ctx.copy());
+ } else {
+ ret.udfList.add(null);
+ }
+ }
+
+ ret.lastQuery = this.lastQuery;
+ ret.udfQuery = this.udfQuery;
+ ret.hasBuiltinAggregation = this.hasBuiltinAggregation;
+
+ return ret;
+ }
}
diff --git
a/server/src/main/java/org/apache/iotdb/db/qp/logical/sys/CreateContinuousQueryOperator.java
b/server/src/main/java/org/apache/iotdb/db/qp/logical/sys/CreateContinuousQueryOperator.java
new file mode 100644
index 0000000..e1516d1
--- /dev/null
+++
b/server/src/main/java/org/apache/iotdb/db/qp/logical/sys/CreateContinuousQueryOperator.java
@@ -0,0 +1,87 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.qp.logical.sys;
+
+import org.apache.iotdb.db.metadata.PartialPath;
+import org.apache.iotdb.db.qp.logical.RootOperator;
+import org.apache.iotdb.db.qp.logical.crud.QueryOperator;
+
+public class CreateContinuousQueryOperator extends RootOperator {
+
+ private String querySql;
+ private QueryOperator queryOperator;
+ private String continuousQueryName;
+ private PartialPath targetPath;
+ private long everyInterval;
+ private long forInterval;
+
+ public CreateContinuousQueryOperator(int tokenIntType) {
+ super(tokenIntType);
+ operatorType = OperatorType.CREATE_CONTINUOUS_QUERY;
+ }
+
+ public void setQuerySql(String querySql) {
+ this.querySql = querySql;
+ }
+
+ public String getQuerySql() {
+ return querySql;
+ }
+
+ public void setContinuousQueryName(String continuousQueryName) {
+ this.continuousQueryName = continuousQueryName;
+ }
+
+ public String getContinuousQueryName() {
+ return continuousQueryName;
+ }
+
+ public void setTargetPath(PartialPath targetPath) {
+ this.targetPath = targetPath;
+ }
+
+ public PartialPath getTargetPath() {
+ return targetPath;
+ }
+
+ public void setEveryInterval(long everyInterval) {
+ this.everyInterval = everyInterval;
+ }
+
+ public long getEveryInterval() {
+ return everyInterval;
+ }
+
+ public void setForInterval(long forInterval) {
+ this.forInterval = forInterval;
+ }
+
+ public long getForInterval() {
+ return forInterval;
+ }
+
+ public void setQueryOperator(QueryOperator queryOperator) {
+ this.queryOperator = queryOperator;
+ }
+
+ public QueryOperator getQueryOperator() {
+ return queryOperator;
+ }
+}
diff --git
a/server/src/main/java/org/apache/iotdb/db/metadata/MetadataOperationType.java
b/server/src/main/java/org/apache/iotdb/db/qp/logical/sys/DropContinuousQueryOperator.java
similarity index 57%
copy from
server/src/main/java/org/apache/iotdb/db/metadata/MetadataOperationType.java
copy to
server/src/main/java/org/apache/iotdb/db/qp/logical/sys/DropContinuousQueryOperator.java
index 466c0a2..34b79aa 100644
---
a/server/src/main/java/org/apache/iotdb/db/metadata/MetadataOperationType.java
+++
b/server/src/main/java/org/apache/iotdb/db/qp/logical/sys/DropContinuousQueryOperator.java
@@ -16,21 +16,25 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.iotdb.db.metadata;
-public class MetadataOperationType {
+package org.apache.iotdb.db.qp.logical.sys;
- private MetadataOperationType() {
- // allowed to do nothing
+import org.apache.iotdb.db.qp.logical.RootOperator;
+
+public class DropContinuousQueryOperator extends RootOperator {
+
+ private String continuousQueryName;
+
+ public DropContinuousQueryOperator(int tokenIntType) {
+ super(tokenIntType);
+ operatorType = OperatorType.DROP_CONTINUOUS_QUERY;
}
- public static final String CREATE_TIMESERIES = "0";
- public static final String DELETE_TIMESERIES = "1";
- public static final String SET_STORAGE_GROUP = "2";
- public static final String SET_TTL = "10";
- public static final String DELETE_STORAGE_GROUP = "11";
- public static final String CREATE_INDEX = "31";
- public static final String DROP_INDEX = "32";
- public static final String CHANGE_OFFSET = "12";
- public static final String CHANGE_ALIAS = "13";
+ public void setContinuousQueryName(String continuousQueryName) {
+ this.continuousQueryName = continuousQueryName;
+ }
+
+ public String getContinuousQueryName() {
+ return continuousQueryName;
+ }
}
diff --git
a/server/src/main/java/org/apache/iotdb/db/metadata/MetadataOperationType.java
b/server/src/main/java/org/apache/iotdb/db/qp/logical/sys/ShowContinuousQueriesOperator.java
similarity index 57%
copy from
server/src/main/java/org/apache/iotdb/db/metadata/MetadataOperationType.java
copy to
server/src/main/java/org/apache/iotdb/db/qp/logical/sys/ShowContinuousQueriesOperator.java
index 466c0a2..c5669ab 100644
---
a/server/src/main/java/org/apache/iotdb/db/metadata/MetadataOperationType.java
+++
b/server/src/main/java/org/apache/iotdb/db/qp/logical/sys/ShowContinuousQueriesOperator.java
@@ -16,21 +16,13 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.iotdb.db.metadata;
-public class MetadataOperationType {
+package org.apache.iotdb.db.qp.logical.sys;
- private MetadataOperationType() {
- // allowed to do nothing
- }
+public class ShowContinuousQueriesOperator extends ShowOperator {
- public static final String CREATE_TIMESERIES = "0";
- public static final String DELETE_TIMESERIES = "1";
- public static final String SET_STORAGE_GROUP = "2";
- public static final String SET_TTL = "10";
- public static final String DELETE_STORAGE_GROUP = "11";
- public static final String CREATE_INDEX = "31";
- public static final String DROP_INDEX = "32";
- public static final String CHANGE_OFFSET = "12";
- public static final String CHANGE_ALIAS = "13";
+ public ShowContinuousQueriesOperator(int tokenIntType) {
+ super(tokenIntType);
+ operatorType = OperatorType.SHOW_CONTINUOUS_QUERIES;
+ }
}
diff --git
a/server/src/main/java/org/apache/iotdb/db/qp/physical/PhysicalPlan.java
b/server/src/main/java/org/apache/iotdb/db/qp/physical/PhysicalPlan.java
index 16719c4..c526d93 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/physical/PhysicalPlan.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/physical/PhysicalPlan.java
@@ -28,6 +28,7 @@ import
org.apache.iotdb.db.qp.physical.crud.InsertMultiTabletPlan;
import org.apache.iotdb.db.qp.physical.crud.InsertRowPlan;
import org.apache.iotdb.db.qp.physical.crud.InsertRowsPlan;
import org.apache.iotdb.db.qp.physical.crud.InsertTabletPlan;
+import org.apache.iotdb.db.qp.physical.sys.*;
import org.apache.iotdb.db.qp.physical.sys.AlterTimeSeriesPlan;
import org.apache.iotdb.db.qp.physical.sys.AuthorPlan;
import org.apache.iotdb.db.qp.physical.sys.ChangeAliasPlan;
@@ -368,6 +369,14 @@ public abstract class PhysicalPlan {
plan = new InsertRowsPlan();
plan.deserialize(buffer);
break;
+ case CREATE_CONTINUOUS_QUERY:
+ plan = new CreateContinuousQueryPlan();
+ plan.deserialize(buffer);
+ break;
+ case DROP_CONTINUOUS_QUERY:
+ plan = new DropContinuousQueryPlan();
+ plan.deserialize(buffer);
+ break;
default:
throw new IOException("unrecognized log type " + type);
}
@@ -412,7 +421,10 @@ public abstract class PhysicalPlan {
BATCH_INSERT_ONE_DEVICE,
MULTI_BATCH_INSERT,
BATCH_INSERT_ROWS,
- SHOW_DEVICES
+ SHOW_DEVICES,
+ CREATE_CONTINUOUS_QUERY,
+ DROP_CONTINUOUS_QUERY,
+ SHOW_CONTINUOUS_QUERIES
}
public long getIndex() {
diff --git
a/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/CreateContinuousQueryPlan.java
b/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/CreateContinuousQueryPlan.java
new file mode 100644
index 0000000..3917143
--- /dev/null
+++
b/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/CreateContinuousQueryPlan.java
@@ -0,0 +1,132 @@
+/*
+ * 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.qp.physical.sys;
+
+import org.apache.iotdb.db.exception.metadata.IllegalPathException;
+import org.apache.iotdb.db.metadata.PartialPath;
+import org.apache.iotdb.db.qp.logical.Operator;
+import org.apache.iotdb.db.qp.logical.crud.QueryOperator;
+import org.apache.iotdb.db.qp.physical.PhysicalPlan;
+import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
+
+import java.nio.ByteBuffer;
+import java.util.*;
+
+public class CreateContinuousQueryPlan extends PhysicalPlan implements
java.io.Serializable {
+
+ private String querySql;
+ private String continuousQueryName;
+ private PartialPath targetPath;
+ private long everyInterval;
+ private long forInterval;
+ private transient QueryOperator queryOperator;
+
+ public CreateContinuousQueryPlan() {
+ super(false, Operator.OperatorType.CREATE_CONTINUOUS_QUERY);
+ }
+
+ public CreateContinuousQueryPlan(
+ String querySql,
+ String continuousQueryName,
+ PartialPath targetPath,
+ long everyInterval,
+ long forInterval,
+ QueryOperator queryOperator) {
+ super(false, Operator.OperatorType.CREATE_CONTINUOUS_QUERY);
+ this.querySql = querySql;
+ this.continuousQueryName = continuousQueryName;
+ this.targetPath = targetPath;
+ this.everyInterval = everyInterval;
+ this.forInterval = forInterval;
+ this.queryOperator = queryOperator;
+ }
+
+ public void setQuerySql(String querySql) {
+ this.querySql = querySql;
+ }
+
+ public String getQuerySql() {
+ return querySql;
+ }
+
+ public void setContinuousQueryName(String continuousQueryName) {
+ this.continuousQueryName = continuousQueryName;
+ }
+
+ public String getContinuousQueryName() {
+ return continuousQueryName;
+ }
+
+ public void setTargetPath(PartialPath targetPath) {
+ this.targetPath = targetPath;
+ }
+
+ public PartialPath getTargetPath() {
+ return targetPath;
+ }
+
+ public void setEveryInterval(long everyInterval) {
+ this.everyInterval = everyInterval;
+ }
+
+ public long getEveryInterval() {
+ return everyInterval;
+ }
+
+ public void setForInterval(long forInterval) {
+ this.forInterval = forInterval;
+ }
+
+ public long getForInterval() {
+ return forInterval;
+ }
+
+ public void setQueryOperator(QueryOperator queryOperator) {
+ this.queryOperator = queryOperator;
+ }
+
+ public QueryOperator getQueryOperator() {
+ return queryOperator;
+ }
+
+ @Override
+ public List<PartialPath> getPaths() {
+ return Collections.emptyList();
+ }
+
+ @Override
+ public void serialize(ByteBuffer buffer) {
+ buffer.put((byte) PhysicalPlanType.CREATE_CONTINUOUS_QUERY.ordinal());
+ ReadWriteIOUtils.write(continuousQueryName, buffer);
+ ReadWriteIOUtils.write(querySql, buffer);
+ ReadWriteIOUtils.write(targetPath.getFullPath(), buffer);
+ buffer.putLong(everyInterval);
+ buffer.putLong(forInterval);
+ }
+
+ @Override
+ public void deserialize(ByteBuffer buffer) throws IllegalPathException {
+ continuousQueryName = ReadWriteIOUtils.readString(buffer);
+ querySql = ReadWriteIOUtils.readString(buffer);
+ targetPath = new PartialPath(ReadWriteIOUtils.readString(buffer));
+ everyInterval = ReadWriteIOUtils.readLong(buffer);
+ forInterval = ReadWriteIOUtils.readLong(buffer);
+ }
+}
diff --git
a/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/DropContinuousQueryPlan.java
b/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/DropContinuousQueryPlan.java
new file mode 100644
index 0000000..1faa458
--- /dev/null
+++
b/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/DropContinuousQueryPlan.java
@@ -0,0 +1,63 @@
+/*
+ * 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.qp.physical.sys;
+
+import org.apache.iotdb.db.metadata.PartialPath;
+import org.apache.iotdb.db.qp.logical.Operator;
+import org.apache.iotdb.db.qp.physical.PhysicalPlan;
+import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
+
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.List;
+
+public class DropContinuousQueryPlan extends PhysicalPlan {
+
+ private String continuousQueryName;
+
+ public DropContinuousQueryPlan() {
+ super(false, Operator.OperatorType.DROP_CONTINUOUS_QUERY);
+ }
+
+ public DropContinuousQueryPlan(String continuousQueryName) {
+ super(false, Operator.OperatorType.DROP_CONTINUOUS_QUERY);
+ this.continuousQueryName = continuousQueryName;
+ }
+
+ @Override
+ public List<PartialPath> getPaths() {
+ return new ArrayList<>();
+ }
+
+ public String getContinuousQueryName() {
+ return continuousQueryName;
+ }
+
+ @Override
+ public void serialize(ByteBuffer buffer) {
+ buffer.put((byte) PhysicalPlanType.DROP_CONTINUOUS_QUERY.ordinal());
+ ReadWriteIOUtils.write(continuousQueryName, buffer);
+ }
+
+ @Override
+ public void deserialize(ByteBuffer buffer) {
+ continuousQueryName = ReadWriteIOUtils.readString(buffer);
+ }
+}
diff --git
a/server/src/main/java/org/apache/iotdb/db/metadata/MetadataOperationType.java
b/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/ShowContinuousQueriesPlan.java
similarity index 57%
copy from
server/src/main/java/org/apache/iotdb/db/metadata/MetadataOperationType.java
copy to
server/src/main/java/org/apache/iotdb/db/qp/physical/sys/ShowContinuousQueriesPlan.java
index 466c0a2..7a4b6ca 100644
---
a/server/src/main/java/org/apache/iotdb/db/metadata/MetadataOperationType.java
+++
b/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/ShowContinuousQueriesPlan.java
@@ -16,21 +16,12 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.iotdb.db.metadata;
-public class MetadataOperationType {
+package org.apache.iotdb.db.qp.physical.sys;
- private MetadataOperationType() {
- // allowed to do nothing
- }
+public class ShowContinuousQueriesPlan extends ShowPlan {
- public static final String CREATE_TIMESERIES = "0";
- public static final String DELETE_TIMESERIES = "1";
- public static final String SET_STORAGE_GROUP = "2";
- public static final String SET_TTL = "10";
- public static final String DELETE_STORAGE_GROUP = "11";
- public static final String CREATE_INDEX = "31";
- public static final String DROP_INDEX = "32";
- public static final String CHANGE_OFFSET = "12";
- public static final String CHANGE_ALIAS = "13";
+ public ShowContinuousQueriesPlan() {
+ super(ShowContentType.CONTINUOUS_QUERY);
+ }
}
diff --git
a/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/ShowPlan.java
b/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/ShowPlan.java
index a77917a..687c9f0 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/ShowPlan.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/ShowPlan.java
@@ -117,6 +117,7 @@ public class ShowPlan extends PhysicalPlan {
COUNT_DEVICES,
COUNT_STORAGE_GROUP,
QUERY_PROCESSLIST,
- TRIGGERS
+ TRIGGERS,
+ CONTINUOUS_QUERY
}
}
diff --git
a/server/src/main/java/org/apache/iotdb/db/qp/sql/IoTDBSqlVisitor.java
b/server/src/main/java/org/apache/iotdb/db/qp/sql/IoTDBSqlVisitor.java
index cfd923b..1ca050e 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/sql/IoTDBSqlVisitor.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/sql/IoTDBSqlVisitor.java
@@ -35,7 +35,7 @@ import org.apache.iotdb.db.qp.logical.crud.InOperator;
import org.apache.iotdb.db.qp.logical.crud.InsertOperator;
import org.apache.iotdb.db.qp.logical.crud.QueryOperator;
import org.apache.iotdb.db.qp.logical.crud.SelectOperator;
-import org.apache.iotdb.db.qp.logical.sys.AlterTimeSeriesOperator;
+import org.apache.iotdb.db.qp.logical.sys.*;
import org.apache.iotdb.db.qp.logical.sys.AlterTimeSeriesOperator.AlterType;
import org.apache.iotdb.db.qp.logical.sys.AuthorOperator;
import org.apache.iotdb.db.qp.logical.sys.AuthorOperator.AuthorType;
@@ -225,6 +225,8 @@ import
org.apache.iotdb.tsfile.file.metadata.enums.TSEncoding;
import org.apache.iotdb.tsfile.utils.Pair;
import org.apache.iotdb.tsfile.utils.StringContainer;
+import org.antlr.v4.runtime.ParserRuleContext;
+import org.antlr.v4.runtime.misc.Interval;
import org.antlr.v4.runtime.tree.TerminalNode;
import java.io.File;
@@ -1035,6 +1037,193 @@ public class IoTDBSqlVisitor extends
SqlBaseBaseVisitor<Operator> {
}
@Override
+ public Operator visitDropContinuousQueryStatement(
+ SqlBaseParser.DropContinuousQueryStatementContext ctx) {
+ DropContinuousQueryOperator operator =
+ new DropContinuousQueryOperator(SQLConstant.TOK_CONTINUOUS_QUERY_DROP);
+ operator.setContinuousQueryName(ctx.continuousQueryName.getText());
+ return operator;
+ }
+
+ @Override
+ public Operator visitShowContinuousQueriesStatement(
+ SqlBaseParser.ShowContinuousQueriesStatementContext ctx) {
+ return new
ShowContinuousQueriesOperator(SQLConstant.TOK_SHOW_CONTINUOUS_QUERIES);
+ }
+
+ @Override
+ public Operator visitCreateContinuousQueryStatement(
+ SqlBaseParser.CreateContinuousQueryStatementContext ctx) {
+ CreateContinuousQueryOperator createContinuousQueryOperator =
+ new
CreateContinuousQueryOperator(SQLConstant.TOK_CONTINUOUS_QUERY_CREATE);
+
+ createContinuousQueryOperator.setQuerySql(ctx.getText());
+
+
createContinuousQueryOperator.setContinuousQueryName(ctx.continuousQueryName.getText());
+
+ if (ctx.resampleClause() != null) {
+ parseResampleClause(ctx.resampleClause(), createContinuousQueryOperator);
+ }
+
+ parseCqSelectIntoClause(ctx.cqSelectIntoClause(),
createContinuousQueryOperator);
+
+ QueryOperator queryOperator =
createContinuousQueryOperator.getQueryOperator();
+
+ StringBuilder sb = new StringBuilder();
+ sb.append("select ");
+ sb.append(ctx.cqSelectIntoClause().selectElements().getText());
+ sb.append(" from ");
+ sb.append(ctx.cqSelectIntoClause().fromClause().prefixPath(0).getText());
+ if (ctx.cqSelectIntoClause().whereClause() != null) {
+ sb.append(" ");
+
sb.append(getFullText(ctx.cqSelectIntoClause().whereClause()).toLowerCase());
+ }
+ sb.append(" group by ([now() - ");
+ String groupByInterval =
ctx.cqSelectIntoClause().cqGroupByTimeClause().DURATION().getText();
+ if (createContinuousQueryOperator.getForInterval() == 0) {
+ sb.append(groupByInterval);
+ } else {
+ List<TerminalNode> durations = ctx.resampleClause().DURATION();
+ sb.append(durations.get(durations.size() - 1).getText());
+ }
+ sb.append(", now()), ");
+ sb.append(groupByInterval);
+ sb.append(")");
+ if (queryOperator.isGroupByLevel()) {
+ sb.append(", level = ");
+ sb.append(queryOperator.getLevel());
+ }
+ createContinuousQueryOperator.setQuerySql(sb.toString());
+
+ if (createContinuousQueryOperator.getEveryInterval() == 0) {
+ if (createContinuousQueryOperator.getForInterval() == 0) {
+
createContinuousQueryOperator.setEveryInterval(queryOperator.getUnit());
+ createContinuousQueryOperator.setForInterval(queryOperator.getUnit());
+ } else {
+ createContinuousQueryOperator.setEveryInterval(
+ createContinuousQueryOperator.getForInterval());
+ }
+ } else {
+ if (createContinuousQueryOperator.getForInterval() == 0) {
+ createContinuousQueryOperator.setForInterval(
+ createContinuousQueryOperator.getEveryInterval());
+ }
+ }
+
+ return createContinuousQueryOperator;
+ }
+
+ private String getFullText(ParserRuleContext context) {
+ if (context.start == null
+ || context.stop == null
+ || context.start.getStartIndex() < 0
+ || context.stop.getStopIndex() < 0) return context.getText();
+
+ return context
+ .start
+ .getInputStream()
+ .getText(Interval.of(context.start.getStartIndex(),
context.stop.getStopIndex()));
+ }
+
+ public void parseResampleClause(
+ SqlBaseParser.ResampleClauseContext ctx, CreateContinuousQueryOperator
operator) {
+
+ if (ctx.DURATION().size() == 1) {
+ if (ctx.EVERY() != null) {
+ operator.setEveryInterval(parseDuration(ctx.DURATION(0).getText()));
+ } else if (ctx.FOR() != null) {
+ operator.setForInterval(parseDuration(ctx.DURATION(0).getText()));
+ }
+ } else if (ctx.DURATION().size() == 2) {
+ operator.setEveryInterval(parseDuration(ctx.DURATION(0).getText()));
+ operator.setForInterval(parseDuration(ctx.DURATION(1).getText()));
+ }
+ }
+
+ public void parseCqSelectIntoClause(
+ SqlBaseParser.CqSelectIntoClauseContext ctx,
+ CreateContinuousQueryOperator createContinuousQueryOperator) {
+
+ QueryOperator queryOperator = new QueryOperator(SQLConstant.TOK_QUERY);
+
+ SelectOperator selectOp = (SelectOperator) visit(ctx.selectElements());
+ if (selectOp.getSuffixPaths().size() > 1) {
+ throw new SQLParserException("Cq Select Into currently does not support
multiple suffixes.");
+ }
+ queryOperator.setSelectOperator(selectOp);
+
+ FromOperator fromOp = (FromOperator) visit(ctx.fromClause());
+ if (fromOp.getPrefixPaths().size() > 1) {
+ throw new SQLParserException("Cq Select Into currently does not support
multiple prefixes.");
+ }
+ queryOperator.setFromOperator(fromOp);
+
+ if (ctx.whereClause() != null) {
+ Operator operator = visit(ctx.whereClause());
+ if (operator instanceof FilterOperator) {
+ FilterOperator whereOp = (FilterOperator) operator;
+ queryOperator.setFilterOperator(whereOp.getChildren().get(0));
+ }
+ }
+
+ parseCqGroupByTimeClause(ctx.cqGroupByTimeClause(), queryOperator);
+
+ int fromLen =
queryOperator.getFromOperator().getPrefixPaths().get(0).getNodeLength();
+ if (queryOperator.getLevel() >= fromLen) {
+ throw new SQLParserException(
+ "Cq Select Into: Level should not exceed the <from_prefix> length.");
+ }
+
+ PartialPath targetPath = null;
+ if (ctx.fullPath() != null) {
+ targetPath = parseFullPath(ctx.fullPath());
+ } else if (ctx.suffixPath() != null) {
+ List<String> targetNodes = new ArrayList<>();
+ int trueLevel = queryOperator.getLevel();
+ if (trueLevel == -1) {
+ trueLevel = fromLen - 1;
+ }
+ for (int i = 0; i <= trueLevel; i++) {
+ targetNodes.add("${" + i + "}");
+ }
+ targetNodes.add(ctx.suffixPath().getText());
+ targetPath = new PartialPath(targetNodes.toArray(new String[0]));
+ }
+
+ createContinuousQueryOperator.setTargetPath(targetPath);
+ createContinuousQueryOperator.setQueryOperator(queryOperator);
+ }
+
+ public void parseCqGroupByTimeClause(
+ SqlBaseParser.CqGroupByTimeClauseContext ctx, QueryOperator
queryOperator) {
+ queryOperator.setGroupByTime(true);
+ queryOperator.setLeftCRightO(true);
+
+ queryOperator.setUnit(parseDuration(ctx.DURATION().getText()));
+ queryOperator.setSlidingStep(queryOperator.getUnit());
+
+ if (ctx.LEVEL() != null && ctx.INT() != null) {
+ queryOperator.setGroupByLevel(true);
+ queryOperator.setLevel(Integer.parseInt(ctx.INT().getText()));
+ }
+ }
+
+ @Override
+ public Operator visitResampleClause(SqlBaseParser.ResampleClauseContext ctx)
{
+ return visitChildren(ctx);
+ }
+
+ @Override
+ public Operator
visitCqSelectIntoClause(SqlBaseParser.CqSelectIntoClauseContext ctx) {
+ return visitChildren(ctx);
+ }
+
+ @Override
+ public Operator
visitCqGroupByTimeClause(SqlBaseParser.CqGroupByTimeClauseContext ctx) {
+ return visitChildren(ctx);
+ }
+
+ @Override
public Operator visitAggregationElement(AggregationElementContext ctx) {
SelectOperator selectOp = new SelectOperator(SQLConstant.TOK_SELECT,
zoneId);
@@ -1077,7 +1266,7 @@ public class IoTDBSqlVisitor extends
SqlBaseBaseVisitor<Operator> {
SelectOperator selectOp = new SelectOperator(SQLConstant.TOK_SELECT,
zoneId);
selectOp.setLastQuery();
LastClauseContext lastClauseContext = ctx.lastClause();
- if (lastClauseContext.asClause().size() != 0) {
+ if (!lastClauseContext.asClause().isEmpty()) {
parseAsClause(lastClauseContext.asClause(), selectOp);
} else {
List<SuffixPathContext> suffixPaths = lastClauseContext.suffixPath();
diff --git
a/server/src/main/java/org/apache/iotdb/db/qp/strategy/PhysicalGenerator.java
b/server/src/main/java/org/apache/iotdb/db/qp/strategy/PhysicalGenerator.java
index dd514d3..8514b3d 100644
---
a/server/src/main/java/org/apache/iotdb/db/qp/strategy/PhysicalGenerator.java
+++
b/server/src/main/java/org/apache/iotdb/db/qp/strategy/PhysicalGenerator.java
@@ -33,6 +33,8 @@ import org.apache.iotdb.db.qp.logical.crud.DeleteDataOperator;
import org.apache.iotdb.db.qp.logical.crud.FilterOperator;
import org.apache.iotdb.db.qp.logical.crud.InsertOperator;
import org.apache.iotdb.db.qp.logical.crud.QueryOperator;
+import org.apache.iotdb.db.qp.logical.sys.CreateContinuousQueryOperator;
+import org.apache.iotdb.db.qp.logical.sys.DropContinuousQueryOperator;
import org.apache.iotdb.db.qp.logical.sys.AlterTimeSeriesOperator;
import org.apache.iotdb.db.qp.logical.sys.AuthorOperator;
import org.apache.iotdb.db.qp.logical.sys.CountOperator;
@@ -83,6 +85,9 @@ 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.UDTFPlan;
+import org.apache.iotdb.db.qp.physical.sys.CreateContinuousQueryPlan;
+import org.apache.iotdb.db.qp.physical.sys.DropContinuousQueryPlan;
+import org.apache.iotdb.db.qp.physical.sys.ShowContinuousQueriesPlan;
import org.apache.iotdb.db.qp.physical.sys.AlterTimeSeriesPlan;
import org.apache.iotdb.db.qp.physical.sys.AuthorPlan;
import org.apache.iotdb.db.qp.physical.sys.ClearCachePlan;
@@ -391,6 +396,21 @@ public class PhysicalGenerator {
return new StartTriggerPlan(((StartTriggerOperator)
operator).getTriggerName());
case STOP_TRIGGER:
return new StopTriggerPlan(((StopTriggerOperator)
operator).getTriggerName());
+ case CREATE_CONTINUOUS_QUERY:
+ CreateContinuousQueryOperator createContinuousQueryOperator =
+ (CreateContinuousQueryOperator) operator;
+ return new CreateContinuousQueryPlan(
+ createContinuousQueryOperator.getQuerySql(),
+ createContinuousQueryOperator.getContinuousQueryName(),
+ createContinuousQueryOperator.getTargetPath(),
+ createContinuousQueryOperator.getEveryInterval(),
+ createContinuousQueryOperator.getForInterval(),
+ createContinuousQueryOperator.getQueryOperator());
+ case DROP_CONTINUOUS_QUERY:
+ return new DropContinuousQueryPlan(
+ ((DropContinuousQueryOperator) operator).getContinuousQueryName());
+ case SHOW_CONTINUOUS_QUERIES:
+ return new ShowContinuousQueriesPlan();
default:
throw new LogicalOperatorException(operator.getType().toString(), "");
}
diff --git
a/server/src/main/java/org/apache/iotdb/db/query/dataset/ShowContinuousQueriesResult.java
b/server/src/main/java/org/apache/iotdb/db/query/dataset/ShowContinuousQueriesResult.java
new file mode 100644
index 0000000..4fb831b
--- /dev/null
+++
b/server/src/main/java/org/apache/iotdb/db/query/dataset/ShowContinuousQueriesResult.java
@@ -0,0 +1,84 @@
+/*
+ * 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.query.dataset;
+
+import org.apache.iotdb.db.metadata.PartialPath;
+
+public class ShowContinuousQueriesResult extends ShowResult {
+
+ private String querySql;
+ private String continuousQueryName;
+ private PartialPath targetPath;
+ private long everyInterval;
+ private long forInterval;
+
+ public ShowContinuousQueriesResult(
+ String querySql,
+ String continuousQueryName,
+ PartialPath targetPath,
+ long everyInterval,
+ long forInterval) {
+ this.querySql = querySql;
+ this.continuousQueryName = continuousQueryName;
+ this.targetPath = targetPath;
+ this.everyInterval = everyInterval;
+ this.forInterval = forInterval;
+ }
+
+ public String getQuerySql() {
+ return querySql;
+ }
+
+ public void setQuerySql(String querySql) {
+ this.querySql = querySql;
+ }
+
+ public String getContinuousQueryName() {
+ return continuousQueryName;
+ }
+
+ public void setContinuousQueryName(String continuousQueryName) {
+ this.continuousQueryName = continuousQueryName;
+ }
+
+ public PartialPath getTargetPath() {
+ return targetPath;
+ }
+
+ public void setTargetPath(PartialPath targetPath) {
+ this.targetPath = targetPath;
+ }
+
+ public long getEveryInterval() {
+ return everyInterval;
+ }
+
+ public void setEveryInterval(long everyInterval) {
+ this.everyInterval = everyInterval;
+ }
+
+ public long getForInterval() {
+ return forInterval;
+ }
+
+ public void setForInterval(long forInterval) {
+ this.forInterval = forInterval;
+ }
+}
diff --git
a/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/GroupByTimeDataSet.java
b/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/GroupByTimeDataSet.java
index 0c61934..8048c55 100644
---
a/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/GroupByTimeDataSet.java
+++
b/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/GroupByTimeDataSet.java
@@ -19,7 +19,9 @@
package org.apache.iotdb.db.query.dataset.groupby;
+import org.apache.iotdb.db.exception.metadata.IllegalPathException;
import org.apache.iotdb.db.exception.query.QueryProcessException;
+import org.apache.iotdb.db.metadata.PartialPath;
import org.apache.iotdb.db.qp.physical.crud.GroupByTimePlan;
import org.apache.iotdb.db.query.aggregation.AggregateResult;
import org.apache.iotdb.db.query.context.QueryContext;
@@ -80,8 +82,13 @@ public class GroupByTimeDataSet extends QueryDataSet {
this.dataTypes = new ArrayList<>();
this.paths = new ArrayList<>();
- for (int i = 0; i < finalPaths.size(); i++) {
- this.dataTypes.add(TSDataType.INT64);
+ for (Map.Entry<String, AggregateResult> entry : finalPaths.entrySet()) {
+ try {
+ this.paths.add(new PartialPath(entry.getKey()));
+ } catch (IllegalPathException e) {
+ e.printStackTrace();
+ }
+ this.dataTypes.add(entry.getValue().getResultDataType());
}
}
diff --git
a/server/src/main/java/org/apache/iotdb/db/query/udf/core/context/UDFContext.java
b/server/src/main/java/org/apache/iotdb/db/query/udf/core/context/UDFContext.java
index 00e294e..2c95e7f 100644
---
a/server/src/main/java/org/apache/iotdb/db/query/udf/core/context/UDFContext.java
+++
b/server/src/main/java/org/apache/iotdb/db/query/udf/core/context/UDFContext.java
@@ -19,6 +19,7 @@
package org.apache.iotdb.db.query.udf.core.context;
+import org.apache.iotdb.db.exception.metadata.IllegalPathException;
import org.apache.iotdb.db.exception.metadata.MetadataException;
import org.apache.iotdb.db.metadata.PartialPath;
import org.apache.iotdb.db.service.IoTDB;
@@ -139,4 +140,31 @@ public class UDFContext {
}
return columnParameterPart;
}
+
+ public UDFContext copy() {
+
+ UDFContext ret = new UDFContext(this.name);
+
+ for (Map.Entry<String, String> attribute : attributes.entrySet()) {
+ ret.attributes.put(attribute.getKey(), attribute.getValue());
+ }
+
+ try {
+ for (PartialPath path : paths) {
+ ret.paths.add(new PartialPath(path.getFullPath()));
+ }
+ } catch (IllegalPathException e) {
+ e.printStackTrace();
+ }
+
+ for (TSDataType type : dataTypes) {
+ ret.dataTypes.add(type);
+ }
+
+ ret.columnParameterPart = this.columnParameterPart;
+
+ ret.column = this.column;
+
+ return ret;
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/service/IoTDB.java
b/server/src/main/java/org/apache/iotdb/db/service/IoTDB.java
index 6a516db..7b2ad93 100644
--- a/server/src/main/java/org/apache/iotdb/db/service/IoTDB.java
+++ b/server/src/main/java/org/apache/iotdb/db/service/IoTDB.java
@@ -23,6 +23,7 @@ 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.cost.statistic.Measurement;
+import org.apache.iotdb.db.cq.ContinuousQueryService;
import org.apache.iotdb.db.engine.StorageEngine;
import org.apache.iotdb.db.engine.cache.CacheHitRatioMonitor;
import org.apache.iotdb.db.engine.compaction.CompactionMergeTaskPoolManager;
@@ -114,6 +115,9 @@ public class IoTDB implements IoTDBMBean {
registerManager.register(UDFClassLoaderManager.getInstance());
registerManager.register(UDFRegistrationService.getInstance());
+ registerManager.register(ContinuousQueryService.getInstance());
+
+ registerManager.register(RPCService.getInstance());
// in cluster mode, RPC service is not enabled.
if (IoTDBDescriptor.getInstance().getConfig().isEnableRpcService()) {
registerManager.register(RPCService.getInstance());
diff --git a/server/src/main/java/org/apache/iotdb/db/service/ServiceType.java
b/server/src/main/java/org/apache/iotdb/db/service/ServiceType.java
index b474b2c..b61ee22 100644
--- a/server/src/main/java/org/apache/iotdb/db/service/ServiceType.java
+++ b/server/src/main/java/org/apache/iotdb/db/service/ServiceType.java
@@ -50,7 +50,8 @@ public enum ServiceType {
FLUSH_SERVICE(
"Flush ServerService",
generateJmxName("org.apache.iotdb.db.engine.pool", "Flush Manager")),
- CLUSTER_MONITOR_SERVICE("Cluster Monitor ServerService", "Cluster Monitor");
+ CLUSTER_MONITOR_SERVICE("Cluster Monitor ServerService", "Cluster Monitor"),
+ CONTINUOUS_QUERY_SERVICE("Continuous Query Service", "Continuous Query
Service");
private final String name;
private final String jmxName;
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 2821aa2..d58b2dc 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
@@ -35,7 +35,6 @@ import
org.apache.iotdb.db.exception.QueryInBatchStatementException;
import org.apache.iotdb.db.exception.StorageEngineException;
import org.apache.iotdb.db.exception.metadata.IllegalPathException;
import org.apache.iotdb.db.exception.metadata.MetadataException;
-import org.apache.iotdb.db.exception.metadata.StorageGroupNotSetException;
import org.apache.iotdb.db.exception.query.QueryProcessException;
import org.apache.iotdb.db.exception.query.QueryTimeoutRuntimeException;
import org.apache.iotdb.db.exception.runtime.SQLParserException;
@@ -1235,7 +1234,7 @@ public class TSServiceImpl implements TSIService.Iface,
ServerContext {
}
private boolean executeNonQuery(PhysicalPlan plan)
- throws QueryProcessException, StorageGroupNotSetException,
StorageEngineException {
+ throws QueryProcessException, MetadataException, StorageEngineException {
if (IoTDBDescriptor.getInstance().getConfig().isReadOnly()) {
throw new QueryProcessException(
"Current system mode is read-only, does not support non-query
operation");
diff --git
a/server/src/main/java/org/apache/iotdb/db/tools/mlog/MLogParser.java
b/server/src/main/java/org/apache/iotdb/db/tools/mlog/MLogParser.java
index 85af420..1be3cef 100644
--- a/server/src/main/java/org/apache/iotdb/db/tools/mlog/MLogParser.java
+++ b/server/src/main/java/org/apache/iotdb/db/tools/mlog/MLogParser.java
@@ -22,14 +22,7 @@ import org.apache.iotdb.db.metadata.MLogTxtWriter;
import org.apache.iotdb.db.metadata.PartialPath;
import org.apache.iotdb.db.metadata.logfile.MLogReader;
import org.apache.iotdb.db.qp.physical.PhysicalPlan;
-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.CreateTimeSeriesPlan;
-import org.apache.iotdb.db.qp.physical.sys.MNodePlan;
-import org.apache.iotdb.db.qp.physical.sys.MeasurementMNodePlan;
-import org.apache.iotdb.db.qp.physical.sys.SetStorageGroupPlan;
-import org.apache.iotdb.db.qp.physical.sys.SetTTLPlan;
-import org.apache.iotdb.db.qp.physical.sys.StorageGroupMNodePlan;
+import org.apache.iotdb.db.qp.physical.sys.*;
import org.apache.commons.cli.CommandLine;
import org.apache.commons.cli.CommandLineParser;
@@ -198,6 +191,12 @@ public class MLogParser {
case MNODE:
mLogTxtWriter.serializeMNode((MNodePlan) plan);
break;
+ case CREATE_CONTINUOUS_QUERY:
+ mLogTxtWriter.createContinuousQuery((CreateContinuousQueryPlan)
plan);
+ break;
+ case DROP_CONTINUOUS_QUERY:
+ mLogTxtWriter.dropContinuousQuery((DropContinuousQueryPlan) plan);
+ break;
default:
logger.warn("unknown plan {}", plan);
}
diff --git
a/server/src/test/java/org/apache/iotdb/db/integration/IoTDBCreateContinuousQueryIT.java
b/server/src/test/java/org/apache/iotdb/db/integration/IoTDBCreateContinuousQueryIT.java
new file mode 100644
index 0000000..a2e38c4
--- /dev/null
+++
b/server/src/test/java/org/apache/iotdb/db/integration/IoTDBCreateContinuousQueryIT.java
@@ -0,0 +1,400 @@
+/*
+ * 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.integration;
+
+import org.apache.iotdb.db.utils.EnvironmentUtils;
+import org.apache.iotdb.jdbc.Config;
+
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.sql.*;
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.List;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.stream.Collectors;
+
+/**
+ * Notice that, all test begins with "IoTDB" is integration test. All test
which will start the
+ * IoTDB server should be defined as integration test.
+ */
+public class IoTDBCreateContinuousQueryIT {
+ private Statement statement;
+ private Connection connection;
+ private final ExecutorService pool = Executors.newCachedThreadPool();
+
+ String[] timeSeriesArray = {
+ "root.ln.wf01.wt01.ws01",
+ "root.ln.wf01.wt01.ws02",
+ "root.ln.wf01.wt02.ws01",
+ "root.ln.wf01.wt02.ws02",
+ "root.ln.wf02.wt01.ws01",
+ "root.ln.wf02.wt01.ws02",
+ "root.ln.wf02.wt02.ws01",
+ "root.ln.wf02.wt02.ws02"
+ };
+
+ @Before
+ public void setUp() throws Exception {
+
+ EnvironmentUtils.envSetUp();
+
+ Class.forName(Config.JDBC_DRIVER_NAME);
+ connection = DriverManager.getConnection("jdbc:iotdb://127.0.0.1:6667/",
"root", "root");
+ statement = connection.createStatement();
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ pool.shutdownNow();
+ statement.close();
+ connection.close();
+ EnvironmentUtils.cleanEnv();
+ }
+
+ @Test
+ public void testCreateContinuousQuery1() throws Exception {
+ for (String timeSeries : timeSeriesArray) {
+ statement.execute(
+ String.format(
+ "create timeseries %s.temperature with
datatype=FLOAT,encoding=RLE", timeSeries));
+ }
+
+ statement.execute(
+ "CREATE CONTINUOUS QUERY cq1 "
+ + "BEGIN SELECT max_value(temperature) INTO temperature_max FROM
root.ln.*.*.* "
+ + "GROUP BY time(10s) END");
+
+ statement.execute(
+ "CREATE CONTINUOUS QUERY cq2 "
+ + "BEGIN SELECT count(temperature) INTO temperature_cnt FROM
root.ln.wf01.*.* "
+ + "WHERE temperature > 80.0 GROUP BY time(10s), level=3 END");
+
+ statement.execute(
+ "CREATE CONTINUOUS QUERY cq3 "
+ + "RESAMPLE EVERY 20s FOR 20s "
+ + "BEGIN SELECT avg(temperature) INTO temperature_avg FROM
root.ln.wf01.*.* "
+ + "GROUP BY time(10s), level=2 END");
+
+ statement.execute("DROP CONTINUOUS QUERY cq1");
+ statement.execute("DROP CONTINUOUS QUERY cq2");
+
+ createCreateContinuousQuery1Tool(new String[] {"cq3"});
+
+ EnvironmentUtils.stopDaemon();
+ setUp();
+
+ createCreateContinuousQuery1Tool(new String[] {"cq3"});
+ }
+
+ private void createCreateContinuousQuery1Tool(String[] continuousQueryArray)
throws SQLException {
+ boolean hasResult = statement.execute("show continuous queries");
+ Assert.assertTrue(hasResult);
+
+ List<String> resultList = new ArrayList<>();
+ try (ResultSet resultSet = statement.getResultSet()) {
+ while (resultSet.next()) {
+ String cq = resultSet.getString("cq name");
+ resultList.add(cq);
+ }
+ }
+ Assert.assertEquals(continuousQueryArray.length, resultList.size());
+
+ List<String> collect =
+ resultList.stream()
+ .sorted(Comparator.comparingInt(e -> e.split("\\.").length))
+ .collect(Collectors.toList());
+
+ for (String s : continuousQueryArray) {
+ Assert.assertTrue(collect.contains(s));
+ }
+ }
+
+ @Test
+ public void testCreateContinuousQuery2() throws Exception {
+
+ for (String timeSeries : timeSeriesArray) {
+ statement.execute(
+ String.format(
+ "create timeseries %s.temperature with
datatype=FLOAT,encoding=RLE", timeSeries));
+ }
+
+ Connection producerConnection =
+ DriverManager.getConnection("jdbc:iotdb://127.0.0.1:6667/", "root",
"root");
+ pool.execute(new Producer(producerConnection));
+
+ statement.execute(
+ "CREATE CONTINUOUS QUERY cq "
+ + "RESAMPLE EVERY 1s FOR 1s "
+ + "BEGIN SELECT avg(temperature) INTO temperature_avg FROM
root.ln.wf01.*.* "
+ + "GROUP BY time(1s), level=2 END");
+
+ Thread.sleep(10000);
+
+ boolean hasResult = statement.execute("select temperature_avg from
root.ln.wf01");
+ Assert.assertTrue(hasResult);
+
+ List<String> resultList = new ArrayList<>();
+ try (ResultSet resultSet = statement.getResultSet()) {
+ while (resultSet.next()) {
+ String timestamp = resultSet.getString("Time");
+ System.out.println(timestamp);
+ resultList.add(timestamp);
+ }
+ }
+ Assert.assertEquals(10000 / 1000, resultList.size());
+ }
+
+ @Test
+ public void testCreateContinuousQuery3() throws Exception {
+
+ for (String timeSeries : timeSeriesArray) {
+ statement.execute(
+ String.format(
+ "create timeseries %s.temperature with
datatype=FLOAT,encoding=RLE", timeSeries));
+ }
+
+ Connection producerConnection =
+ DriverManager.getConnection("jdbc:iotdb://127.0.0.1:6667/", "root",
"root");
+ pool.execute(new Producer(producerConnection));
+
+ statement.execute(
+ "CREATE CONTINUOUS QUERY cq "
+ + "RESAMPLE EVERY 20s "
+ + "BEGIN SELECT avg(temperature) INTO temperature_avg FROM
root.ln.wf01.*.* "
+ + "GROUP BY time(1s), level=2 END");
+
+ Thread.sleep(40000);
+
+ boolean hasResult = statement.execute("select temperature_avg from
root.ln.wf01");
+ Assert.assertTrue(hasResult);
+
+ List<String> resultList = new ArrayList<>();
+ try (ResultSet resultSet = statement.getResultSet()) {
+ while (resultSet.next()) {
+ String timestamp = resultSet.getString("Time");
+ System.out.println(timestamp);
+ resultList.add(timestamp);
+ }
+ }
+ Assert.assertEquals(40000 / 1000, resultList.size());
+ }
+
+ @Test
+ public void testCreateContinuousQuery4() throws Exception {
+
+ for (String timeSeries : timeSeriesArray) {
+ statement.execute(
+ String.format(
+ "create timeseries %s.temperature with
datatype=FLOAT,encoding=RLE", timeSeries));
+ }
+
+ Connection producerConnection =
+ DriverManager.getConnection("jdbc:iotdb://127.0.0.1:6667/", "root",
"root");
+ pool.execute(new Producer(producerConnection));
+
+ statement.execute(
+ "CREATE CONTINUOUS QUERY cq "
+ + "BEGIN SELECT avg(temperature) INTO temperature_avg FROM
root.ln.wf01.*.* "
+ + "GROUP BY time(1s), level=2 END");
+
+ Thread.sleep(10000);
+
+ boolean hasResult = statement.execute("select temperature_avg from
root.ln.wf01");
+ Assert.assertTrue(hasResult);
+
+ List<String> resultList = new ArrayList<>();
+ try (ResultSet resultSet = statement.getResultSet()) {
+ while (resultSet.next()) {
+ String timestamp = resultSet.getString("Time");
+ System.out.println(timestamp);
+ resultList.add(timestamp);
+ }
+ }
+ Assert.assertEquals(10000 / 1000, resultList.size());
+ }
+
+ @Test
+ public void testCreateContinuousQuery5() throws Exception {
+
+ for (String timeSeries : timeSeriesArray) {
+ statement.execute(
+ String.format(
+ "create timeseries %s.temperature with
datatype=FLOAT,encoding=RLE", timeSeries));
+ }
+
+ Connection producerConnection =
+ DriverManager.getConnection("jdbc:iotdb://127.0.0.1:6667/", "root",
"root");
+ pool.execute(new Producer(producerConnection));
+
+ statement.execute(
+ "CREATE CONTINUOUS QUERY cq "
+ + "BEGIN SELECT avg(temperature) INTO temperature_avg FROM
root.ln.wf01.*.* WHERE temperature < 0 "
+ + "GROUP BY time(1s), level=2 END");
+
+ Thread.sleep(10000);
+
+ boolean hasResult = statement.execute("select temperature_avg from
root.ln.wf01");
+ Assert.assertTrue(hasResult);
+
+ List<String> resultList = new ArrayList<>();
+ try (ResultSet resultSet = statement.getResultSet()) {
+ while (resultSet.next()) {
+ String timestamp = resultSet.getString("Time");
+ System.out.println(timestamp);
+ resultList.add(timestamp);
+ }
+ }
+ Assert.assertEquals(0, resultList.size());
+ }
+
+ @Test
+ public void testCreateContinuousQuery6() throws Exception {
+
+ for (String timeSeries : timeSeriesArray) {
+ statement.execute(
+ String.format(
+ "create timeseries %s.temperature with
datatype=FLOAT,encoding=RLE", timeSeries));
+ }
+
+ Connection producerConnection =
+ DriverManager.getConnection("jdbc:iotdb://127.0.0.1:6667/", "root",
"root");
+ pool.execute(new Producer(producerConnection));
+
+ statement.execute(
+ "CREATE CONTINUOUS QUERY cq "
+ + "BEGIN SELECT count(temperature) INTO temperature_cnt FROM
root.ln.*.*.* "
+ + "GROUP BY time(1s), level=2 END");
+
+ Thread.sleep(10000);
+
+ checkTimeSeries(
+ new String[] {
+ "root.ln.wf01.wt01.ws01.temperature",
+ "root.ln.wf01.wt01.ws02.temperature",
+ "root.ln.wf01.wt02.ws01.temperature",
+ "root.ln.wf01.wt02.ws02.temperature",
+ "root.ln.wf02.wt01.ws01.temperature",
+ "root.ln.wf02.wt01.ws02.temperature",
+ "root.ln.wf02.wt02.ws01.temperature",
+ "root.ln.wf02.wt02.ws02.temperature",
+ "root.ln.wf01.temperature_cnt",
+ "root.ln.wf02.temperature_cnt"
+ });
+ }
+
+ @Test
+ public void testCreateContinuousQuery7() throws Exception {
+
+ statement.execute(
+ "CREATE CONTINUOUS QUERY cq "
+ + "BEGIN SELECT avg(temperature) INTO temperature_avg FROM
root.ln.wf01.*.* "
+ + "GROUP BY time(1s), level=2 END");
+
+ Thread.sleep(10000);
+
+ for (String timeSeries : timeSeriesArray) {
+ statement.execute(
+ String.format(
+ "create timeseries %s.temperature with
datatype=FLOAT,encoding=RLE", timeSeries));
+ }
+
+ Connection producerConnection =
+ DriverManager.getConnection("jdbc:iotdb://127.0.0.1:6667/", "root",
"root");
+ pool.execute(new Producer(producerConnection));
+
+ Thread.sleep(10000);
+
+ boolean hasResult = statement.execute("select temperature_avg from
root.ln.wf01");
+ Assert.assertTrue(hasResult);
+
+ List<String> resultList = new ArrayList<>();
+ try (ResultSet resultSet = statement.getResultSet()) {
+ while (resultSet.next()) {
+ String timestamp = resultSet.getString("Time");
+ System.out.println(timestamp);
+ resultList.add(timestamp);
+ }
+ }
+ Assert.assertEquals(10000 / 1000, resultList.size());
+ }
+
+ private void checkTimeSeries(String[] timeSeriesArray) throws SQLException {
+ boolean hasResult = statement.execute("show timeseries");
+ Assert.assertTrue(hasResult);
+
+ List<String> resultList = new ArrayList<>();
+ try (ResultSet resultSet = statement.getResultSet()) {
+ while (resultSet.next()) {
+ String timeseries = resultSet.getString("timeseries");
+ resultList.add(timeseries);
+ }
+ }
+ Assert.assertEquals(timeSeriesArray.length, resultList.size());
+
+ List<String> collect =
+ resultList.stream()
+ .sorted(Comparator.comparingInt(e -> e.split("\\.").length))
+ .collect(Collectors.toList());
+
+ for (String s : timeSeriesArray) {
+ Assert.assertTrue(collect.contains(s));
+ }
+ }
+
+ class Producer implements Runnable {
+ private final Statement producerStatement;
+ private final Connection producerConnection;
+
+ public Producer(Connection producerConnection) throws SQLException {
+ this.producerConnection = producerConnection;
+ this.producerStatement = producerConnection.createStatement();
+ }
+
+ protected void finalize() throws SQLException {
+ producerStatement.close();
+ producerConnection.close();
+ }
+
+ @Override
+ public void run() {
+
+ while (!Thread.currentThread().isInterrupted()) {
+ try {
+ Thread.sleep(100);
+ for (String timeSeries : timeSeriesArray) {
+ this.producerStatement.execute(
+ String.format(
+ "insert into %s(timestamp, temperature) values(now(),
%.3f)",
+ timeSeries, 200 * Math.random()));
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ } catch (SQLException throwables) {
+ throwables.printStackTrace();
+ }
+ }
+ }
+ }
+}
diff --git
a/server/src/test/java/org/apache/iotdb/db/qp/physical/PhysicalPlanTest.java
b/server/src/test/java/org/apache/iotdb/db/qp/physical/PhysicalPlanTest.java
index e6aaa95..c4ac722 100644
--- a/server/src/test/java/org/apache/iotdb/db/qp/physical/PhysicalPlanTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/qp/physical/PhysicalPlanTest.java
@@ -37,6 +37,9 @@ 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.UDTFPlan;
+import org.apache.iotdb.db.qp.physical.sys.CreateContinuousQueryPlan;
+import org.apache.iotdb.db.qp.physical.sys.DropContinuousQueryPlan;
+import org.apache.iotdb.db.qp.physical.sys.ShowContinuousQueriesPlan;
import org.apache.iotdb.db.qp.physical.sys.AuthorPlan;
import org.apache.iotdb.db.qp.physical.sys.CreateFunctionPlan;
import org.apache.iotdb.db.qp.physical.sys.CreateTimeSeriesPlan;
@@ -1215,4 +1218,125 @@ public class PhysicalPlanTest {
Assert.assertTrue(plan.isQuery());
Assert.assertEquals("root.sg1.d1.s1", plan.getPath().getFullPath());
}
+
+ @Test
+ public void testCreateCQ1() throws QueryProcessException {
+ String sql =
+ "CREATE CONTINUOUS QUERY cq1 BEGIN SELECT max_value(temperature) INTO
temperature_max FROM root.ln.*.*.* GROUP BY time(10s) END";
+
+ CreateContinuousQueryPlan plan =
+ (CreateContinuousQueryPlan) processor.parseSQLToPhysicalPlan(sql);
+ Assert.assertFalse(plan.isQuery());
+ Assert.assertEquals("cq1", plan.getContinuousQueryName());
+ Assert.assertEquals(10000, plan.getEveryInterval());
+ Assert.assertEquals(10000, plan.getForInterval());
+ Assert.assertEquals(
+ "${0}.${1}.${2}.${3}.${4}.temperature_max",
plan.getTargetPath().getFullPath());
+ Assert.assertEquals(
+ "select max_value(temperature) from root.ln.*.*.* group by ([now() -
10s, now()), 10s)",
+ plan.getQuerySql());
+ }
+
+ @Test
+ public void testCreateCQ2() throws QueryProcessException {
+ String sql =
+ "CREATE CONTINUOUS QUERY cq1 BEGIN SELECT max_value(temperature) INTO
temperature_max FROM root.ln.*.*.* GROUP BY time(10s), level = 3 END";
+
+ CreateContinuousQueryPlan plan =
+ (CreateContinuousQueryPlan) processor.parseSQLToPhysicalPlan(sql);
+ Assert.assertFalse(plan.isQuery());
+ Assert.assertEquals("cq1", plan.getContinuousQueryName());
+ Assert.assertEquals(10000, plan.getEveryInterval());
+ Assert.assertEquals(10000, plan.getForInterval());
+ Assert.assertEquals("${0}.${1}.${2}.${3}.temperature_max",
plan.getTargetPath().getFullPath());
+ Assert.assertEquals(
+ "select max_value(temperature) from root.ln.*.*.* group by ([now() -
10s, now()), 10s), level = 3",
+ plan.getQuerySql());
+ }
+
+ @Test
+ public void testCreateCQ3() throws QueryProcessException {
+ String sql =
+ "CREATE CONTINUOUS QUERY cq1 RESAMPLE EVERY 20s BEGIN SELECT
max_value(temperature) INTO temperature_max FROM root.ln.*.*.* GROUP BY
time(10s), level = 3 END";
+
+ CreateContinuousQueryPlan plan =
+ (CreateContinuousQueryPlan) processor.parseSQLToPhysicalPlan(sql);
+ Assert.assertFalse(plan.isQuery());
+ Assert.assertEquals("cq1", plan.getContinuousQueryName());
+ Assert.assertEquals(20000, plan.getEveryInterval());
+ Assert.assertEquals(20000, plan.getForInterval());
+ Assert.assertEquals("${0}.${1}.${2}.${3}.temperature_max",
plan.getTargetPath().getFullPath());
+ Assert.assertEquals(
+ "select max_value(temperature) from root.ln.*.*.* group by ([now() -
10s, now()), 10s), level = 3",
+ plan.getQuerySql());
+ }
+
+ @Test
+ public void testCreateCQ4() throws QueryProcessException {
+ String sql =
+ "CREATE CONTINUOUS QUERY cq1 RESAMPLE EVERY 20s FOR 10s BEGIN SELECT
max_value(temperature) INTO temperature_max FROM root.ln.*.*.* GROUP BY
time(10s), level = 3 END";
+
+ CreateContinuousQueryPlan plan =
+ (CreateContinuousQueryPlan) processor.parseSQLToPhysicalPlan(sql);
+ Assert.assertFalse(plan.isQuery());
+ Assert.assertEquals("cq1", plan.getContinuousQueryName());
+ Assert.assertEquals(20000, plan.getEveryInterval());
+ Assert.assertEquals(10000, plan.getForInterval());
+ Assert.assertEquals("${0}.${1}.${2}.${3}.temperature_max",
plan.getTargetPath().getFullPath());
+ Assert.assertEquals(
+ "select max_value(temperature) from root.ln.*.*.* group by ([now() -
10s, now()), 10s), level = 3",
+ plan.getQuerySql());
+ }
+
+ @Test
+ public void testCreateCQ5() throws QueryProcessException {
+ String sql =
+ "CREATE CONTINUOUS QUERY cq1 RESAMPLE FOR 20s BEGIN SELECT
max_value(temperature) INTO temperature_max FROM root.ln.*.*.* GROUP BY
time(10s), level = 3 END";
+
+ CreateContinuousQueryPlan plan =
+ (CreateContinuousQueryPlan) processor.parseSQLToPhysicalPlan(sql);
+ Assert.assertFalse(plan.isQuery());
+ Assert.assertEquals("cq1", plan.getContinuousQueryName());
+ Assert.assertEquals(20000, plan.getEveryInterval());
+ Assert.assertEquals(20000, plan.getForInterval());
+ Assert.assertEquals("${0}.${1}.${2}.${3}.temperature_max",
plan.getTargetPath().getFullPath());
+ Assert.assertEquals(
+ "select max_value(temperature) from root.ln.*.*.* group by ([now() -
20s, now()), 10s), level = 3",
+ plan.getQuerySql());
+ }
+
+ @Test
+ public void testCreateCQ6() throws QueryProcessException {
+ String sql =
+ "CREATE CONTINUOUS QUERY cq1 RESAMPLE FOR 20s BEGIN SELECT
max_value(temperature) INTO temperature_max FROM root.ln.*.*.* WHERE
temperature > 50 and temperature < 100 GROUP BY time(10s), level = 3 END";
+
+ CreateContinuousQueryPlan plan =
+ (CreateContinuousQueryPlan) processor.parseSQLToPhysicalPlan(sql);
+ Assert.assertFalse(plan.isQuery());
+ Assert.assertEquals("cq1", plan.getContinuousQueryName());
+ Assert.assertEquals(20000, plan.getEveryInterval());
+ Assert.assertEquals(20000, plan.getForInterval());
+ Assert.assertEquals("${0}.${1}.${2}.${3}.temperature_max",
plan.getTargetPath().getFullPath());
+ Assert.assertEquals(
+ "select max_value(temperature) from root.ln.*.*.* where temperature >
50 and temperature < 100 group by ([now() - 20s, now()), 10s), level = 3",
+ plan.getQuerySql());
+ }
+
+ @Test
+ public void testDropCQ() throws QueryProcessException {
+ String sql = "DROP CONTINUOUS QUERY cq1";
+
+ DropContinuousQueryPlan plan = (DropContinuousQueryPlan)
processor.parseSQLToPhysicalPlan(sql);
+ Assert.assertFalse(plan.isQuery());
+ Assert.assertEquals("cq1", plan.getContinuousQueryName());
+ }
+
+ @Test
+ public void testShowCQs() throws QueryProcessException {
+ String sql = "SHOW CONTINUOUS QUERIES";
+
+ ShowContinuousQueriesPlan plan =
+ (ShowContinuousQueriesPlan) processor.parseSQLToPhysicalPlan(sql);
+ Assert.assertTrue(plan.isQuery());
+ }
}
diff --git a/service-rpc/src/main/java/org/apache/iotdb/rpc/TSStatusCode.java
b/service-rpc/src/main/java/org/apache/iotdb/rpc/TSStatusCode.java
index f2fd3c1..ed695b1 100644
--- a/service-rpc/src/main/java/org/apache/iotdb/rpc/TSStatusCode.java
+++ b/service-rpc/src/main/java/org/apache/iotdb/rpc/TSStatusCode.java
@@ -46,6 +46,8 @@ public enum TSStatusCode {
LOAD_FILE_ERROR(316),
STORAGE_GROUP_NOT_READY(317),
ILLEGAL_PARAMETER(318),
+ CONTINUOUS_QUERY_ALREADY_EXIST(319),
+ CONTINUOUS_QUERY_NOT_EXIST(320),
EXECUTE_STATEMENT_ERROR(400),
SQL_PARSE_ERROR(401),
