You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ja...@apache.org on 2022/10/26 07:41:15 UTC
[iotdb] branch IOTDB-4619 updated: [IOTDB-4620] Implemation of CQ on DataNode (#7734)
This is an automated email from the ASF dual-hosted git repository.
jackietien pushed a commit to branch IOTDB-4619
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/IOTDB-4619 by this push:
new 1ed166b534 [IOTDB-4620] Implemation of CQ on DataNode (#7734)
1ed166b534 is described below
commit 1ed166b5346c2d6d1776a30806c6d8da84afefd4
Author: liuminghui233 <36...@users.noreply.github.com>
AuthorDate: Wed Oct 26 15:41:10 2022 +0800
[IOTDB-4620] Implemation of CQ on DataNode (#7734)
---
.../org/apache/iotdb/db/qp/sql/IoTDBSqlParser.g4 | 34 ++--
.../antlr4/org/apache/iotdb/db/qp/sql/SqlLexer.g4 | 24 ++-
.../org/apache/iotdb/db/mpp/common/QueryId.java | 2 +
.../db/mpp/common/header/ColumnHeaderConstant.java | 11 ++
.../db/mpp/common/header/DatasetHeaderFactory.java | 4 +
.../iotdb/db/mpp/plan/analyze/AnalyzeVisitor.java | 11 +-
.../apache/iotdb/db/mpp/plan/analyze/Analyzer.java | 22 ++-
.../db/mpp/plan/analyze/ExpressionAnalyzer.java | 26 +++
.../plan/execution/config/ConfigTaskVisitor.java | 24 +++
.../config/executor/ClusterConfigTaskExecutor.java | 81 +++++++++
.../config/executor/IConfigTaskExecutor.java | 8 +
.../executor/StandaloneConfigTaskExecutor.java | 46 +++++
.../config/metadata/CreateContinuousQueryTask.java | 42 +++++
.../config/metadata/DropContinuousQueryTask.java | 42 +++++
.../config/metadata/ShowContinuousQueriesTask.java | 75 ++++++++
.../iotdb/db/mpp/plan/parser/ASTVisitor.java | 111 ++++++++++--
.../db/mpp/plan/statement/StatementVisitor.java | 19 +-
.../db/mpp/plan/statement/crud/QueryStatement.java | 40 +++--
.../metadata/CreateContinuousQueryStatement.java | 192 +++++++++++++++++++++
.../metadata/DropContinuousQueryStatement.java | 60 +++++++
.../metadata/ShowContinuousQueriesStatement.java | 52 ++++++
.../apache/iotdb/db/qp/sql/IoTDBSqlVisitor.java | 56 +++---
22 files changed, 906 insertions(+), 76 deletions(-)
diff --git a/antlr/src/main/antlr4/org/apache/iotdb/db/qp/sql/IoTDBSqlParser.g4 b/antlr/src/main/antlr4/org/apache/iotdb/db/qp/sql/IoTDBSqlParser.g4
index 5ef759a569..3d897fc513 100644
--- a/antlr/src/main/antlr4/org/apache/iotdb/db/qp/sql/IoTDBSqlParser.g4
+++ b/antlr/src/main/antlr4/org/apache/iotdb/db/qp/sql/IoTDBSqlParser.g4
@@ -152,20 +152,24 @@ triggerAttribute
// Create Continuous Query
createContinuousQuery
- : CREATE (CONTINUOUS QUERY | CQ) continuousQueryName=identifier resampleClause? cqSelectIntoClause
+ : CREATE (CONTINUOUS QUERY | CQ) cqId=identifier
+ resampleClause?
+ timeoutPolicyClause?
+ BEGIN
+ selectStatement
+ END
;
-cqSelectIntoClause
- : BEGIN selectClause INTO intoPath fromClause cqGroupByTimeClause END
- ;
-
-cqGroupByTimeClause
- : GROUP BY TIME LR_BRACKET DURATION_LITERAL RR_BRACKET
- (COMMA LEVEL operator_eq INTEGER_LITERAL (COMMA INTEGER_LITERAL)*)?
+resampleClause
+ : RESAMPLE
+ (EVERY everyInterval=DURATION_LITERAL)?
+ (FOR DURATION_LITERAL)?
+ (BOUNDARY boundaryTime=dateExpression)?
+ (RANGE startTimeOffset=DURATION_LITERAL (COMMA endTimeOffset=DURATION_LITERAL))?
;
-resampleClause
- : RESAMPLE (EVERY DURATION_LITERAL)? (FOR DURATION_LITERAL)? (BOUNDARY dateExpression)?
+timeoutPolicyClause
+ : TIMEOUT POLICY (BLOCKED | DISCARD)
;
// Alter Timeseries
@@ -223,7 +227,7 @@ dropTrigger
// Drop Continuous Query
dropContinuousQuery
- : DROP (CONTINUOUS QUERY|CQ) continuousQueryName=identifier
+ : DROP (CONTINUOUS QUERY|CQ) cqId=identifier
;
// Drop Schema Template
@@ -472,15 +476,15 @@ sortKey
;
groupByTimeClause
- : GROUP BY LR_BRACKET timeRange COMMA DURATION_LITERAL (COMMA DURATION_LITERAL)? fillClause? RR_BRACKET
- | GROUP BY LR_BRACKET timeRange COMMA DURATION_LITERAL (COMMA DURATION_LITERAL)? RR_BRACKET
+ : GROUP BY TIME? LR_BRACKET (timeRange COMMA)? interval=DURATION_LITERAL (COMMA step=DURATION_LITERAL)? fillClause? RR_BRACKET
+ | GROUP BY TIME? LR_BRACKET (timeRange COMMA)? interval=DURATION_LITERAL (COMMA step=DURATION_LITERAL)? RR_BRACKET
COMMA LEVEL operator_eq INTEGER_LITERAL (COMMA INTEGER_LITERAL)* fillClause?
- | GROUP BY LR_BRACKET timeRange COMMA DURATION_LITERAL (COMMA DURATION_LITERAL)? RR_BRACKET
+ | GROUP BY TIME? LR_BRACKET (timeRange COMMA)? interval=DURATION_LITERAL (COMMA step=DURATION_LITERAL)? RR_BRACKET
COMMA TAGS LR_BRACKET identifier (COMMA identifier)* RR_BRACKET
;
groupByFillClause
- : GROUP BY LR_BRACKET timeRange COMMA DURATION_LITERAL (COMMA DURATION_LITERAL)? RR_BRACKET
+ : GROUP BY TIME? LR_BRACKET (timeRange COMMA)? interval=DURATION_LITERAL (COMMA step=DURATION_LITERAL)? RR_BRACKET
fillClause
;
diff --git a/antlr/src/main/antlr4/org/apache/iotdb/db/qp/sql/SqlLexer.g4 b/antlr/src/main/antlr4/org/apache/iotdb/db/qp/sql/SqlLexer.g4
index f595fcf9b1..badfb2a81f 100644
--- a/antlr/src/main/antlr4/org/apache/iotdb/db/qp/sql/SqlLexer.g4
+++ b/antlr/src/main/antlr4/org/apache/iotdb/db/qp/sql/SqlLexer.g4
@@ -93,6 +93,10 @@ BEGIN
: B E G I N
;
+BLOCKED
+ : B L O C K E D
+ ;
+
BOUNDARY
: B O U N D A R Y
;
@@ -194,6 +198,10 @@ DISABLE
: D I S A B L E
;
+DISCARD
+ : D I S C A R D
+ ;
+
DROP
: D R O P
;
@@ -407,6 +415,10 @@ PIPESINKTYPE
: P I P E S I N K T Y P E
;
+POLICY
+ : P O L I C Y
+ ;
+
PREVIOUS
: P R E V I O U S
;
@@ -439,8 +451,8 @@ QUERY
: Q U E R Y
;
-RUNNING
- : R U N N I N G
+RANGE
+ : R A N G E
;
READONLY
@@ -487,6 +499,10 @@ ROOT
: R O O T
;
+RUNNING
+ : R U N N I N G
+ ;
+
SCHEMA
: S C H E M A
;
@@ -571,6 +587,10 @@ TIME
: T I M E
;
+TIMEOUT
+ : T I M E O U T
+ ;
+
TIMESERIES
: T I M E S E R I E S
;
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/common/QueryId.java b/server/src/main/java/org/apache/iotdb/db/mpp/common/QueryId.java
index d927fe3040..be9462696f 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/common/QueryId.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/common/QueryId.java
@@ -33,6 +33,8 @@ import static java.util.Objects.requireNonNull;
public class QueryId {
+ public static QueryId mockQueryId = QueryId.valueOf("mock_query_id");
+
private final String id;
private int nextPlanNodeIndex;
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/common/header/ColumnHeaderConstant.java b/server/src/main/java/org/apache/iotdb/db/mpp/common/header/ColumnHeaderConstant.java
index 2ebb1705a7..7924503897 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/common/header/ColumnHeaderConstant.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/common/header/ColumnHeaderConstant.java
@@ -121,6 +121,11 @@ public class ColumnHeaderConstant {
public static final String COLUMN_TARGET_TIMESERIES = "target timeseries";
public static final String COLUMN_WRITTEN = "written";
+ // column names for show cq
+ public static final String COLUMN_CQ_ID = "cq id";
+ public static final String COLUMN_QUERY = "query";
+ public static final String COLUMN_STATE = "state";
+
public static final List<ColumnHeader> lastQueryColumnHeaders =
ImmutableList.of(
new ColumnHeader(COLUMN_TIMESERIES, TSDataType.TEXT),
@@ -303,4 +308,10 @@ public class ColumnHeaderConstant {
public static final List<ColumnHeader> getSeriesSlotListColumnHeaders =
ImmutableList.of(new ColumnHeader(COLUMN_SERIES_SLOT_ID, TSDataType.INT32));
+
+ public static final List<ColumnHeader> showContinuousQueriesColumnHeaders =
+ ImmutableList.of(
+ new ColumnHeader(COLUMN_CQ_ID, TSDataType.TEXT),
+ new ColumnHeader(COLUMN_QUERY, TSDataType.TEXT),
+ new ColumnHeader(COLUMN_STATE, TSDataType.TEXT));
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/common/header/DatasetHeaderFactory.java b/server/src/main/java/org/apache/iotdb/db/mpp/common/header/DatasetHeaderFactory.java
index f64300890b..020f6426e1 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/common/header/DatasetHeaderFactory.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/common/header/DatasetHeaderFactory.java
@@ -146,4 +146,8 @@ public class DatasetHeaderFactory {
? new DatasetHeader(ColumnHeaderConstant.selectIntoAlignByDeviceColumnHeaders, true)
: new DatasetHeader(ColumnHeaderConstant.selectIntoColumnHeaders, true);
}
+
+ public static DatasetHeader getShowContinuousQueriesHeader() {
+ return new DatasetHeader(ColumnHeaderConstant.showContinuousQueriesColumnHeaders, true);
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/AnalyzeVisitor.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/AnalyzeVisitor.java
index 925d5558ec..320b5025cb 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/AnalyzeVisitor.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/AnalyzeVisitor.java
@@ -29,7 +29,6 @@ import org.apache.iotdb.commons.partition.SchemaPartition;
import org.apache.iotdb.commons.path.MeasurementPath;
import org.apache.iotdb.commons.path.PartialPath;
import org.apache.iotdb.commons.path.PathPatternTree;
-import org.apache.iotdb.db.conf.IoTDBConfig;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
import org.apache.iotdb.db.engine.storagegroup.TsFileResourceStatus;
@@ -157,15 +156,11 @@ import static org.apache.iotdb.db.mpp.plan.analyze.SelectIntoUtils.constructTarg
public class AnalyzeVisitor extends StatementVisitor<Analysis, MPPQueryContext> {
private static final Logger logger = LoggerFactory.getLogger(Analyzer.class);
- private static final IoTDBConfig config = IoTDBDescriptor.getInstance().getConfig();
private final IPartitionFetcher partitionFetcher;
private final ISchemaFetcher schemaFetcher;
- private final MPPQueryContext context;
- public AnalyzeVisitor(
- IPartitionFetcher partitionFetcher, ISchemaFetcher schemaFetcher, MPPQueryContext context) {
- this.context = context;
+ public AnalyzeVisitor(IPartitionFetcher partitionFetcher, ISchemaFetcher schemaFetcher) {
this.partitionFetcher = partitionFetcher;
this.schemaFetcher = schemaFetcher;
}
@@ -1052,6 +1047,10 @@ public class AnalyzeVisitor extends StatementVisitor<Analysis, MPPQueryContext>
&& queryStatement.getResultTimeOrder() == Ordering.DESC) {
throw new SemanticException("Group by month doesn't support order by time desc now.");
}
+ if (!queryStatement.isCqQueryBody()
+ && (groupByTimeComponent.getStartTime() == 0 && groupByTimeComponent.getEndTime() == 0)) {
+ throw new SemanticException("The query time range should be specified in the GROUP BY TIME clause.");
+ }
analysis.setGroupByTimeParameter(new GroupByTimeParameter(groupByTimeComponent));
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/Analyzer.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/Analyzer.java
index 9697bf29e2..010b491e55 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/Analyzer.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/Analyzer.java
@@ -19,9 +19,12 @@
package org.apache.iotdb.db.mpp.plan.analyze;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.mpp.common.MPPQueryContext;
import org.apache.iotdb.db.mpp.plan.statement.Statement;
+import static org.apache.iotdb.db.mpp.common.QueryId.mockQueryId;
+
/** Analyze the statement and generate Analysis. */
public class Analyzer {
private final MPPQueryContext context;
@@ -37,6 +40,23 @@ public class Analyzer {
}
public Analysis analyze(Statement statement) {
- return new AnalyzeVisitor(partitionFetcher, schemaFetcher, context).process(statement, context);
+ return new AnalyzeVisitor(partitionFetcher, schemaFetcher).process(statement, context);
+ }
+
+ public static void validate(Statement statement) {
+ MPPQueryContext context = new MPPQueryContext(mockQueryId);
+
+ IPartitionFetcher partitionFetcher;
+ ISchemaFetcher schemaFetcher;
+ if (IoTDBDescriptor.getInstance().getConfig().isClusterMode()) {
+ partitionFetcher = ClusterPartitionFetcher.getInstance();
+ schemaFetcher = ClusterSchemaFetcher.getInstance();
+ } else {
+ partitionFetcher = StandalonePartitionFetcher.getInstance();
+ schemaFetcher = StandaloneSchemaFetcher.getInstance();
+ }
+
+ Analyzer analyzer = new Analyzer(context, partitionFetcher, schemaFetcher);
+ analyzer.analyze(statement);
}
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/ExpressionAnalyzer.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/ExpressionAnalyzer.java
index 0f9e31dae6..688f881ae9 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/ExpressionAnalyzer.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/ExpressionAnalyzer.java
@@ -904,6 +904,32 @@ public class ExpressionAnalyzer {
}
}
+ public static boolean checkIfTimeFilterExist(Expression predicate) {
+ if (predicate instanceof TernaryExpression) {
+ return checkIfTimeFilterExist(((TernaryExpression) predicate).getFirstExpression())
+ || checkIfTimeFilterExist(((TernaryExpression) predicate).getSecondExpression())
+ || checkIfTimeFilterExist(((TernaryExpression) predicate).getThirdExpression());
+ } else if (predicate instanceof BinaryExpression) {
+ return checkIfTimeFilterExist(((BinaryExpression) predicate).getLeftExpression())
+ || checkIfTimeFilterExist(((BinaryExpression) predicate).getRightExpression());
+ } else if (predicate instanceof UnaryExpression) {
+ return checkIfTimeFilterExist(((UnaryExpression) predicate).getExpression());
+ } else if (predicate instanceof FunctionExpression) {
+ boolean timeFilterExist = false;
+ for (Expression childExpression : predicate.getExpressions()) {
+ timeFilterExist = timeFilterExist || checkIfTimeFilterExist(childExpression);
+ }
+ return timeFilterExist;
+ } else if (predicate instanceof TimeSeriesOperand || predicate instanceof ConstantOperand) {
+ return false;
+ } else if (predicate instanceof TimestampOperand) {
+ return true;
+ } else {
+ throw new IllegalArgumentException(
+ "unsupported expression type: " + predicate.getExpressionType());
+ }
+ }
+
/**
* Search for subexpressions that can be queried natively, including all time series.
*
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/ConfigTaskVisitor.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/ConfigTaskVisitor.java
index db163bbac9..c97ea6758d 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/ConfigTaskVisitor.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/ConfigTaskVisitor.java
@@ -20,10 +20,12 @@
package org.apache.iotdb.db.mpp.plan.execution.config;
import org.apache.iotdb.db.mpp.plan.execution.config.metadata.CountStorageGroupTask;
+import org.apache.iotdb.db.mpp.plan.execution.config.metadata.CreateContinuousQueryTask;
import org.apache.iotdb.db.mpp.plan.execution.config.metadata.CreateFunctionTask;
import org.apache.iotdb.db.mpp.plan.execution.config.metadata.CreateTriggerTask;
import org.apache.iotdb.db.mpp.plan.execution.config.metadata.DeleteStorageGroupTask;
import org.apache.iotdb.db.mpp.plan.execution.config.metadata.DeleteTimeSeriesTask;
+import org.apache.iotdb.db.mpp.plan.execution.config.metadata.DropContinuousQueryTask;
import org.apache.iotdb.db.mpp.plan.execution.config.metadata.DropFunctionTask;
import org.apache.iotdb.db.mpp.plan.execution.config.metadata.DropTriggerTask;
import org.apache.iotdb.db.mpp.plan.execution.config.metadata.GetRegionIdTask;
@@ -33,6 +35,7 @@ import org.apache.iotdb.db.mpp.plan.execution.config.metadata.SetStorageGroupTas
import org.apache.iotdb.db.mpp.plan.execution.config.metadata.SetTTLTask;
import org.apache.iotdb.db.mpp.plan.execution.config.metadata.ShowClusterTask;
import org.apache.iotdb.db.mpp.plan.execution.config.metadata.ShowConfigNodesTask;
+import org.apache.iotdb.db.mpp.plan.execution.config.metadata.ShowContinuousQueriesTask;
import org.apache.iotdb.db.mpp.plan.execution.config.metadata.ShowDataNodesTask;
import org.apache.iotdb.db.mpp.plan.execution.config.metadata.ShowFunctionsTask;
import org.apache.iotdb.db.mpp.plan.execution.config.metadata.ShowRegionTask;
@@ -64,10 +67,12 @@ import org.apache.iotdb.db.mpp.plan.statement.Statement;
import org.apache.iotdb.db.mpp.plan.statement.StatementNode;
import org.apache.iotdb.db.mpp.plan.statement.StatementVisitor;
import org.apache.iotdb.db.mpp.plan.statement.metadata.CountStorageGroupStatement;
+import org.apache.iotdb.db.mpp.plan.statement.metadata.CreateContinuousQueryStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.CreateFunctionStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.CreateTriggerStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.DeleteStorageGroupStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.DeleteTimeSeriesStatement;
+import org.apache.iotdb.db.mpp.plan.statement.metadata.DropContinuousQueryStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.DropFunctionStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.DropTriggerStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.GetRegionIdStatement;
@@ -77,6 +82,7 @@ import org.apache.iotdb.db.mpp.plan.statement.metadata.SetStorageGroupStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.SetTTLStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.ShowClusterStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.ShowConfigNodesStatement;
+import org.apache.iotdb.db.mpp.plan.statement.metadata.ShowContinuousQueriesStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.ShowDataNodesStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.ShowFunctionsStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.ShowRegionStatement;
@@ -352,6 +358,24 @@ public class ConfigTaskVisitor
return new GetTimeSlotListTask(getTimeSlotListStatement);
}
+ @Override
+ public IConfigTask visitCreateContinuousQuery(
+ CreateContinuousQueryStatement createContinuousQueryStatement, TaskContext context) {
+ return new CreateContinuousQueryTask(createContinuousQueryStatement);
+ }
+
+ @Override
+ public IConfigTask visitDropContinuousQuery(
+ DropContinuousQueryStatement dropContinuousQueryStatement, TaskContext context) {
+ return new DropContinuousQueryTask(dropContinuousQueryStatement);
+ }
+
+ @Override
+ public IConfigTask visitShowContinuousQueries(
+ ShowContinuousQueriesStatement showContinuousQueriesStatement, TaskContext context) {
+ return new ShowContinuousQueriesTask();
+ }
+
public static class TaskContext {
private final String queryId;
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/executor/ClusterConfigTaskExecutor.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/executor/ClusterConfigTaskExecutor.java
index 4cab93e460..f2903a073c 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/executor/ClusterConfigTaskExecutor.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/executor/ClusterConfigTaskExecutor.java
@@ -32,12 +32,14 @@ import org.apache.iotdb.commons.path.PartialPath;
import org.apache.iotdb.commons.path.PathPatternTree;
import org.apache.iotdb.commons.trigger.service.TriggerExecutableManager;
import org.apache.iotdb.confignode.rpc.thrift.TCountStorageGroupResp;
+import org.apache.iotdb.confignode.rpc.thrift.TCreateCQReq;
import org.apache.iotdb.confignode.rpc.thrift.TCreateFunctionReq;
import org.apache.iotdb.confignode.rpc.thrift.TCreatePipeReq;
import org.apache.iotdb.confignode.rpc.thrift.TCreateTriggerReq;
import org.apache.iotdb.confignode.rpc.thrift.TDeactivateSchemaTemplateReq;
import org.apache.iotdb.confignode.rpc.thrift.TDeleteStorageGroupsReq;
import org.apache.iotdb.confignode.rpc.thrift.TDeleteTimeSeriesReq;
+import org.apache.iotdb.confignode.rpc.thrift.TDropCQReq;
import org.apache.iotdb.confignode.rpc.thrift.TDropFunctionReq;
import org.apache.iotdb.confignode.rpc.thrift.TDropPipeSinkReq;
import org.apache.iotdb.confignode.rpc.thrift.TDropTriggerReq;
@@ -53,6 +55,7 @@ import org.apache.iotdb.confignode.rpc.thrift.TGetTimeSlotListResp;
import org.apache.iotdb.confignode.rpc.thrift.TGetTriggerTableResp;
import org.apache.iotdb.confignode.rpc.thrift.TPipeSinkInfo;
import org.apache.iotdb.confignode.rpc.thrift.TSetStorageGroupReq;
+import org.apache.iotdb.confignode.rpc.thrift.TShowCQResp;
import org.apache.iotdb.confignode.rpc.thrift.TShowClusterResp;
import org.apache.iotdb.confignode.rpc.thrift.TShowConfigNodesResp;
import org.apache.iotdb.confignode.rpc.thrift.TShowDataNodesResp;
@@ -69,6 +72,7 @@ import org.apache.iotdb.db.client.DataNodeClientPoolFactory;
import org.apache.iotdb.db.localconfignode.LocalConfigNode;
import org.apache.iotdb.db.metadata.template.ClusterTemplateManager;
import org.apache.iotdb.db.metadata.template.Template;
+import org.apache.iotdb.db.mpp.plan.analyze.Analyzer;
import org.apache.iotdb.db.mpp.plan.execution.config.ConfigTaskResult;
import org.apache.iotdb.db.mpp.plan.execution.config.metadata.CountStorageGroupTask;
import org.apache.iotdb.db.mpp.plan.execution.config.metadata.GetRegionIdTask;
@@ -77,6 +81,7 @@ import org.apache.iotdb.db.mpp.plan.execution.config.metadata.GetTimeSlotListTas
import org.apache.iotdb.db.mpp.plan.execution.config.metadata.SetStorageGroupTask;
import org.apache.iotdb.db.mpp.plan.execution.config.metadata.ShowClusterTask;
import org.apache.iotdb.db.mpp.plan.execution.config.metadata.ShowConfigNodesTask;
+import org.apache.iotdb.db.mpp.plan.execution.config.metadata.ShowContinuousQueriesTask;
import org.apache.iotdb.db.mpp.plan.execution.config.metadata.ShowDataNodesTask;
import org.apache.iotdb.db.mpp.plan.execution.config.metadata.ShowRegionTask;
import org.apache.iotdb.db.mpp.plan.execution.config.metadata.ShowStorageGroupTask;
@@ -88,6 +93,7 @@ import org.apache.iotdb.db.mpp.plan.execution.config.metadata.template.ShowSchem
import org.apache.iotdb.db.mpp.plan.execution.config.sys.sync.ShowPipeSinkTask;
import org.apache.iotdb.db.mpp.plan.execution.config.sys.sync.ShowPipeTask;
import org.apache.iotdb.db.mpp.plan.statement.metadata.CountStorageGroupStatement;
+import org.apache.iotdb.db.mpp.plan.statement.metadata.CreateContinuousQueryStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.CreateTriggerStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.DeleteStorageGroupStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.DeleteTimeSeriesStatement;
@@ -1167,4 +1173,79 @@ public class ClusterConfigTaskExecutor implements IConfigTaskExecutor {
GetTimeSlotListTask.buildTSBlock(resp, future);
return future;
}
+
+ @Override
+ public SettableFuture<ConfigTaskResult> createContinuousQuery(
+ CreateContinuousQueryStatement createContinuousQueryStatement) {
+ createContinuousQueryStatement.semanticCheck();
+ Analyzer.validate(createContinuousQueryStatement.getQueryBodyStatement());
+
+ SettableFuture<ConfigTaskResult> future = SettableFuture.create();
+ try (ConfigNodeClient client =
+ CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.partitionRegionId)) {
+ TCreateCQReq tCreateCQReq =
+ new TCreateCQReq(
+ createContinuousQueryStatement.getCqId(),
+ createContinuousQueryStatement.getEveryInterval(),
+ createContinuousQueryStatement.getBoundaryTime(),
+ createContinuousQueryStatement.getStartTimeOffset(),
+ createContinuousQueryStatement.getEndTimeOffset(),
+ createContinuousQueryStatement.getTimeoutPolicy().getType(),
+ createContinuousQueryStatement.getQueryBody(),
+ createContinuousQueryStatement.getSql(),
+ createContinuousQueryStatement.getZoneId());
+ final TSStatus executionStatus = client.createCQ(tCreateCQReq);
+ if (TSStatusCode.SUCCESS_STATUS.getStatusCode() != executionStatus.getCode()) {
+ LOGGER.error(
+ "[{}] Failed to create continuous query {}. TSStatus is {}",
+ executionStatus,
+ createContinuousQueryStatement.getCqId(),
+ executionStatus.message);
+ future.setException(new IoTDBException(executionStatus.message, executionStatus.code));
+ } else {
+ future.set(new ConfigTaskResult(TSStatusCode.SUCCESS_STATUS));
+ }
+ } catch (TException | IOException e) {
+ future.setException(e);
+ }
+ return future;
+ }
+
+ @Override
+ public SettableFuture<ConfigTaskResult> dropContinuousQuery(String cqId) {
+ SettableFuture<ConfigTaskResult> future = SettableFuture.create();
+ try (ConfigNodeClient client =
+ CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.partitionRegionId)) {
+ final TSStatus executionStatus = client.dropCQ(new TDropCQReq(cqId));
+ if (TSStatusCode.SUCCESS_STATUS.getStatusCode() != executionStatus.getCode()) {
+ LOGGER.error("[{}] Failed to drop continuous query {}.", executionStatus, cqId);
+ future.setException(new IoTDBException(executionStatus.message, executionStatus.code));
+ } else {
+ future.set(new ConfigTaskResult(TSStatusCode.SUCCESS_STATUS));
+ }
+ } catch (TException | IOException e) {
+ future.setException(e);
+ }
+ return future;
+ }
+
+ @Override
+ public SettableFuture<ConfigTaskResult> showContinuousQueries() {
+ SettableFuture<ConfigTaskResult> future = SettableFuture.create();
+ try (ConfigNodeClient client =
+ CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.partitionRegionId)) {
+ TShowCQResp showCQResp = client.showCQ();
+ if (showCQResp.getStatus().getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ future.setException(
+ new IoTDBException(showCQResp.getStatus().message, showCQResp.getStatus().code));
+ return future;
+ }
+ // convert cqList and buildTsBlock
+ ShowContinuousQueriesTask.buildTsBlock(showCQResp.getCqList(), future);
+ } catch (TException | IOException e) {
+ future.setException(e);
+ }
+
+ return future;
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/executor/IConfigTaskExecutor.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/executor/IConfigTaskExecutor.java
index 1881fbefd1..97839e9275 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/executor/IConfigTaskExecutor.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/executor/IConfigTaskExecutor.java
@@ -23,6 +23,7 @@ import org.apache.iotdb.common.rpc.thrift.TFlushReq;
import org.apache.iotdb.commons.cluster.NodeStatus;
import org.apache.iotdb.db.mpp.plan.execution.config.ConfigTaskResult;
import org.apache.iotdb.db.mpp.plan.statement.metadata.CountStorageGroupStatement;
+import org.apache.iotdb.db.mpp.plan.statement.metadata.CreateContinuousQueryStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.CreateTriggerStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.DeleteStorageGroupStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.DeleteTimeSeriesStatement;
@@ -145,4 +146,11 @@ public interface IConfigTaskExecutor {
SettableFuture<ConfigTaskResult> getTimeSlotList(
GetTimeSlotListStatement getTimeSlotListStatement);
+
+ SettableFuture<ConfigTaskResult> createContinuousQuery(
+ CreateContinuousQueryStatement createContinuousQueryStatement);
+
+ SettableFuture<ConfigTaskResult> dropContinuousQuery(String cqId);
+
+ SettableFuture<ConfigTaskResult> showContinuousQueries();
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/executor/StandaloneConfigTaskExecutor.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/executor/StandaloneConfigTaskExecutor.java
index c64c8a1469..a6de0f0da2 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/executor/StandaloneConfigTaskExecutor.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/executor/StandaloneConfigTaskExecutor.java
@@ -41,6 +41,7 @@ import org.apache.iotdb.db.mpp.plan.execution.config.metadata.ShowTTLTask;
import org.apache.iotdb.db.mpp.plan.execution.config.sys.sync.ShowPipeSinkTask;
import org.apache.iotdb.db.mpp.plan.execution.config.sys.sync.ShowPipeTask;
import org.apache.iotdb.db.mpp.plan.statement.metadata.CountStorageGroupStatement;
+import org.apache.iotdb.db.mpp.plan.statement.metadata.CreateContinuousQueryStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.CreateTriggerStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.DeleteStorageGroupStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.DeleteTimeSeriesStatement;
@@ -631,4 +632,49 @@ public class StandaloneConfigTaskExecutor implements IConfigTaskExecutor {
TSStatusCode.EXECUTE_STATEMENT_ERROR.getStatusCode()));
return future;
}
+
+ @Override
+ public SettableFuture<ConfigTaskResult> createContinuousQuery(
+ CreateContinuousQueryStatement createContinuousQueryStatement) {
+ SettableFuture<ConfigTaskResult> future = SettableFuture.create();
+ try {
+ // todo: implementation
+ future.set(new ConfigTaskResult(TSStatusCode.SUCCESS_STATUS));
+ } catch (Exception e) {
+ final String message =
+ String.format(
+ "Failed to create continuous query %s, because %s.",
+ createContinuousQueryStatement.getCqId(), e.getMessage());
+ LOGGER.error(message, e);
+ future.setException(
+ new IoTDBException(message, TSStatusCode.EXECUTE_STATEMENT_ERROR.getStatusCode()));
+ }
+ return future;
+ }
+
+ @Override
+ public SettableFuture<ConfigTaskResult> dropContinuousQuery(String cqId) {
+ SettableFuture<ConfigTaskResult> future = SettableFuture.create();
+ try {
+ // todo: implementation
+ future.set(new ConfigTaskResult(TSStatusCode.SUCCESS_STATUS));
+ } catch (Exception e) {
+ final String message =
+ String.format("Failed to continuous query trigger %s, because %s.", cqId, e.getMessage());
+ LOGGER.error(message, e);
+ future.setException(
+ new IoTDBException(message, TSStatusCode.EXECUTE_STATEMENT_ERROR.getStatusCode()));
+ }
+ return future;
+ }
+
+ @Override
+ public SettableFuture<ConfigTaskResult> showContinuousQueries() {
+ SettableFuture<ConfigTaskResult> future = SettableFuture.create();
+ future.setException(
+ new IoTDBException(
+ "Executing show continuous queries in standalone mode is not supported",
+ TSStatusCode.EXECUTE_STATEMENT_ERROR.getStatusCode()));
+ return future;
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/metadata/CreateContinuousQueryTask.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/metadata/CreateContinuousQueryTask.java
new file mode 100644
index 0000000000..060162e731
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/metadata/CreateContinuousQueryTask.java
@@ -0,0 +1,42 @@
+/*
+ * 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.mpp.plan.execution.config.metadata;
+
+import org.apache.iotdb.db.mpp.plan.execution.config.ConfigTaskResult;
+import org.apache.iotdb.db.mpp.plan.execution.config.IConfigTask;
+import org.apache.iotdb.db.mpp.plan.execution.config.executor.IConfigTaskExecutor;
+import org.apache.iotdb.db.mpp.plan.statement.metadata.CreateContinuousQueryStatement;
+
+import com.google.common.util.concurrent.ListenableFuture;
+
+public class CreateContinuousQueryTask implements IConfigTask {
+
+ private final CreateContinuousQueryStatement createContinuousQueryStatement;
+
+ public CreateContinuousQueryTask(CreateContinuousQueryStatement createContinuousQueryStatement) {
+ this.createContinuousQueryStatement = createContinuousQueryStatement;
+ }
+
+ @Override
+ public ListenableFuture<ConfigTaskResult> execute(IConfigTaskExecutor configTaskExecutor)
+ throws InterruptedException {
+ return configTaskExecutor.createContinuousQuery(createContinuousQueryStatement);
+ }
+}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/metadata/DropContinuousQueryTask.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/metadata/DropContinuousQueryTask.java
new file mode 100644
index 0000000000..17a76a56fa
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/metadata/DropContinuousQueryTask.java
@@ -0,0 +1,42 @@
+/*
+ * 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.mpp.plan.execution.config.metadata;
+
+import org.apache.iotdb.db.mpp.plan.execution.config.ConfigTaskResult;
+import org.apache.iotdb.db.mpp.plan.execution.config.IConfigTask;
+import org.apache.iotdb.db.mpp.plan.execution.config.executor.IConfigTaskExecutor;
+import org.apache.iotdb.db.mpp.plan.statement.metadata.DropContinuousQueryStatement;
+
+import com.google.common.util.concurrent.ListenableFuture;
+
+public class DropContinuousQueryTask implements IConfigTask {
+
+ private final String cqId;
+
+ public DropContinuousQueryTask(DropContinuousQueryStatement dropContinuousQueryStatement) {
+ this.cqId = dropContinuousQueryStatement.getCqId();
+ }
+
+ @Override
+ public ListenableFuture<ConfigTaskResult> execute(IConfigTaskExecutor configTaskExecutor)
+ throws InterruptedException {
+ return configTaskExecutor.dropContinuousQuery(cqId);
+ }
+}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/metadata/ShowContinuousQueriesTask.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/metadata/ShowContinuousQueriesTask.java
new file mode 100644
index 0000000000..deb4fb10e6
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/metadata/ShowContinuousQueriesTask.java
@@ -0,0 +1,75 @@
+/*
+ * 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.mpp.plan.execution.config.metadata;
+
+import org.apache.iotdb.commons.cq.CQState;
+import org.apache.iotdb.confignode.rpc.thrift.TCQEntry;
+import org.apache.iotdb.db.mpp.common.header.ColumnHeader;
+import org.apache.iotdb.db.mpp.common.header.ColumnHeaderConstant;
+import org.apache.iotdb.db.mpp.common.header.DatasetHeader;
+import org.apache.iotdb.db.mpp.common.header.DatasetHeaderFactory;
+import org.apache.iotdb.db.mpp.plan.execution.config.ConfigTaskResult;
+import org.apache.iotdb.db.mpp.plan.execution.config.IConfigTask;
+import org.apache.iotdb.db.mpp.plan.execution.config.executor.IConfigTaskExecutor;
+import org.apache.iotdb.rpc.TSStatusCode;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.read.common.block.TsBlockBuilder;
+import org.apache.iotdb.tsfile.read.common.block.column.ColumnBuilder;
+import org.apache.iotdb.tsfile.read.common.block.column.TimeColumnBuilder;
+import org.apache.iotdb.tsfile.utils.Binary;
+
+import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.SettableFuture;
+
+import java.util.List;
+import java.util.stream.Collectors;
+
+public class ShowContinuousQueriesTask implements IConfigTask {
+
+ @Override
+ public ListenableFuture<ConfigTaskResult> execute(IConfigTaskExecutor configTaskExecutor)
+ throws InterruptedException {
+ return configTaskExecutor.showContinuousQueries();
+ }
+
+ public static void buildTsBlock(List<TCQEntry> cqList, SettableFuture<ConfigTaskResult> future) {
+ List<TSDataType> outputDataTypes =
+ ColumnHeaderConstant.showContinuousQueriesColumnHeaders.stream()
+ .map(ColumnHeader::getColumnType)
+ .collect(Collectors.toList());
+ TsBlockBuilder builder = new TsBlockBuilder(outputDataTypes);
+
+ if (cqList != null && !cqList.isEmpty()) {
+ TimeColumnBuilder timeColumnBuilder = builder.getTimeColumnBuilder();
+ ColumnBuilder[] columnBuilders = builder.getValueColumnBuilders();
+ for (TCQEntry cqEntry : cqList) {
+ timeColumnBuilder.writeLong(0);
+ columnBuilders[0].writeBinary(Binary.valueOf(cqEntry.getCqId()));
+ columnBuilders[1].writeBinary(Binary.valueOf(cqEntry.getSql()));
+ columnBuilders[2].writeBinary(
+ Binary.valueOf(CQState.values()[cqEntry.getState()].toString()));
+ builder.declarePosition();
+ }
+ }
+
+ DatasetHeader datasetHeader = DatasetHeaderFactory.getShowContinuousQueriesHeader();
+ future.set(new ConfigTaskResult(TSStatusCode.SUCCESS_STATUS, builder.build(), datasetHeader));
+ }
+}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/parser/ASTVisitor.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/parser/ASTVisitor.java
index 14bf81bd82..2e6cc90895 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/parser/ASTVisitor.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/parser/ASTVisitor.java
@@ -25,6 +25,7 @@ import org.apache.iotdb.common.rpc.thrift.TTimePartitionSlot;
import org.apache.iotdb.commons.auth.entity.PrivilegeType;
import org.apache.iotdb.commons.cluster.NodeStatus;
import org.apache.iotdb.commons.conf.IoTDBConstant;
+import org.apache.iotdb.commons.cq.TimeoutPolicy;
import org.apache.iotdb.commons.exception.IllegalPathException;
import org.apache.iotdb.commons.path.PartialPath;
import org.apache.iotdb.commons.utils.PathUtils;
@@ -98,11 +99,13 @@ import org.apache.iotdb.db.mpp.plan.statement.metadata.CountNodesStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.CountStorageGroupStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.CountTimeSeriesStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.CreateAlignedTimeSeriesStatement;
+import org.apache.iotdb.db.mpp.plan.statement.metadata.CreateContinuousQueryStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.CreateFunctionStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.CreateTimeSeriesStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.CreateTriggerStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.DeleteStorageGroupStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.DeleteTimeSeriesStatement;
+import org.apache.iotdb.db.mpp.plan.statement.metadata.DropContinuousQueryStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.DropFunctionStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.DropTriggerStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.GetRegionIdStatement;
@@ -114,6 +117,7 @@ import org.apache.iotdb.db.mpp.plan.statement.metadata.ShowChildNodesStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.ShowChildPathsStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.ShowClusterStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.ShowConfigNodesStatement;
+import org.apache.iotdb.db.mpp.plan.statement.metadata.ShowContinuousQueriesStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.ShowDataNodesStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.ShowDevicesStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.ShowFunctionsStatement;
@@ -775,11 +779,13 @@ public class ASTVisitor extends IoTDBSqlParserBaseVisitor<Statement> {
attributes);
}
+ // Drop Trigger =====================================================================
@Override
public Statement visitDropTrigger(IoTDBSqlParser.DropTriggerContext ctx) {
return new DropTriggerStatement(parseIdentifier(ctx.triggerName.getText()));
}
+ // Show Trigger =====================================================================
@Override
public Statement visitShowTriggers(IoTDBSqlParser.ShowTriggersContext ctx) {
return new ShowTriggersStatement();
@@ -805,6 +811,83 @@ public class ASTVisitor extends IoTDBSqlParserBaseVisitor<Statement> {
}
}
+ // Create CQ =====================================================================
+ @Override
+ public Statement visitCreateContinuousQuery(IoTDBSqlParser.CreateContinuousQueryContext ctx) {
+ CreateContinuousQueryStatement statement = new CreateContinuousQueryStatement();
+ statement.setSql(ctx.getText());
+ statement.setCqId(parseIdentifier(ctx.cqId.getText()));
+
+ statement.setQueryBody(ctx.selectStatement().getText());
+ QueryStatement queryBodyStatement =
+ (QueryStatement) visitSelectStatement(ctx.selectStatement());
+ queryBodyStatement.setCqQueryBody(true);
+ statement.setQueryBodyStatement(queryBodyStatement);
+
+ if (ctx.resampleClause() != null) {
+ parseResampleClause(ctx.resampleClause(), statement);
+ }
+
+ if (ctx.timeoutPolicyClause() != null) {
+ parseTimeoutPolicyClause(ctx.timeoutPolicyClause(), statement);
+ }
+
+ return statement;
+ }
+
+ private void parseResampleClause(
+ IoTDBSqlParser.ResampleClauseContext ctx, CreateContinuousQueryStatement statement) {
+ if (ctx.FOR() != null) {
+ throw new SemanticException("CQ: The syntax of CREATE CQ statement has changed from v0.14");
+ }
+
+ if (ctx.EVERY() != null) {
+ statement.setEveryInterval(
+ DateTimeUtils.convertDurationStrToLong(ctx.everyInterval.getText()));
+ } else {
+ QueryStatement queryStatement = statement.getQueryBodyStatement();
+ if (!queryStatement.isGroupByTime()) {
+ throw new SemanticException(
+ "CQ: At least one of the parameters `every_interval` and `group_by_interval` needs to be specified.");
+ }
+ statement.setEveryInterval(queryStatement.getGroupByTimeComponent().getInterval());
+ }
+
+ if (ctx.BOUNDARY() != null) {
+ statement.setBoundaryTime(parseDateExpression(ctx.boundaryTime));
+ }
+
+ if (ctx.RANGE() != null) {
+ statement.setStartTimeOffset(
+ DateTimeUtils.convertDurationStrToLong(ctx.startTimeOffset.getText()));
+ if (ctx.endTimeOffset != null) {
+ statement.setEndTimeOffset(
+ DateTimeUtils.convertDurationStrToLong(ctx.endTimeOffset.getText()));
+ }
+ } else {
+ statement.setStartTimeOffset(statement.getEveryInterval());
+ }
+ }
+
+ private void parseTimeoutPolicyClause(
+ IoTDBSqlParser.TimeoutPolicyClauseContext ctx, CreateContinuousQueryStatement statement) {
+ if (ctx.DISCARD() != null) {
+ statement.setTimeoutPolicy(TimeoutPolicy.DISCARD);
+ }
+ }
+
+ // Drop CQ =====================================================================
+ @Override
+ public Statement visitDropContinuousQuery(IoTDBSqlParser.DropContinuousQueryContext ctx) {
+ return new DropContinuousQueryStatement(parseIdentifier(ctx.cqId.getText()));
+ }
+
+ // Show CQs =====================================================================
+ @Override
+ public Statement visitShowContinuousQueries(IoTDBSqlParser.ShowContinuousQueriesContext ctx) {
+ return new ShowContinuousQueriesStatement();
+ }
+
/** Data Manipulation Language (DML) */
// Select Statement ========================================================================
@@ -952,23 +1035,23 @@ public class ASTVisitor extends IoTDBSqlParserBaseVisitor<Statement> {
GroupByTimeComponent groupByTimeComponent = new GroupByTimeComponent();
// parse time range
- parseTimeRange(ctx.timeRange(), groupByTimeComponent);
- groupByTimeComponent.setLeftCRightO(ctx.timeRange().LS_BRACKET() != null);
+ if (ctx.timeRange() != null) {
+ parseTimeRange(ctx.timeRange(), groupByTimeComponent);
+ groupByTimeComponent.setLeftCRightO(ctx.timeRange().LS_BRACKET() != null);
+ }
// parse time interval
groupByTimeComponent.setInterval(
- parseTimeIntervalOrSlidingStep(
- ctx.DURATION_LITERAL(0).getText(), true, groupByTimeComponent));
+ parseTimeIntervalOrSlidingStep(ctx.interval.getText(), true, groupByTimeComponent));
if (groupByTimeComponent.getInterval() <= 0) {
throw new SemanticException(
"The second parameter time interval should be a positive integer.");
}
// parse sliding step
- if (ctx.DURATION_LITERAL().size() == 2) {
+ if (ctx.step != null) {
groupByTimeComponent.setSlidingStep(
- parseTimeIntervalOrSlidingStep(
- ctx.DURATION_LITERAL(1).getText(), false, groupByTimeComponent));
+ parseTimeIntervalOrSlidingStep(ctx.step.getText(), false, groupByTimeComponent));
} else {
groupByTimeComponent.setSlidingStep(groupByTimeComponent.getInterval());
groupByTimeComponent.setSlidingStepByMonth(groupByTimeComponent.isIntervalByMonth());
@@ -1006,23 +1089,23 @@ public class ASTVisitor extends IoTDBSqlParserBaseVisitor<Statement> {
GroupByTimeComponent groupByTimeComponent = new GroupByTimeComponent();
// parse time range
- parseTimeRange(ctx.timeRange(), groupByTimeComponent);
- groupByTimeComponent.setLeftCRightO(ctx.timeRange().LS_BRACKET() != null);
+ if (ctx.timeRange() != null) {
+ parseTimeRange(ctx.timeRange(), groupByTimeComponent);
+ groupByTimeComponent.setLeftCRightO(ctx.timeRange().LS_BRACKET() != null);
+ }
// parse time interval
groupByTimeComponent.setInterval(
- parseTimeIntervalOrSlidingStep(
- ctx.DURATION_LITERAL(0).getText(), true, groupByTimeComponent));
+ parseTimeIntervalOrSlidingStep(ctx.interval.getText(), true, groupByTimeComponent));
if (groupByTimeComponent.getInterval() <= 0) {
throw new SemanticException(
"The second parameter time interval should be a positive integer.");
}
// parse sliding step
- if (ctx.DURATION_LITERAL().size() == 2) {
+ if (ctx.step != null) {
groupByTimeComponent.setSlidingStep(
- parseTimeIntervalOrSlidingStep(
- ctx.DURATION_LITERAL(1).getText(), false, groupByTimeComponent));
+ parseTimeIntervalOrSlidingStep(ctx.step.getText(), false, groupByTimeComponent));
} else {
groupByTimeComponent.setSlidingStep(groupByTimeComponent.getInterval());
groupByTimeComponent.setSlidingStepByMonth(groupByTimeComponent.isIntervalByMonth());
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/statement/StatementVisitor.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/statement/StatementVisitor.java
index 0c74684373..abc53975be 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/statement/StatementVisitor.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/statement/StatementVisitor.java
@@ -37,12 +37,14 @@ import org.apache.iotdb.db.mpp.plan.statement.metadata.CountNodesStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.CountStorageGroupStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.CountTimeSeriesStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.CreateAlignedTimeSeriesStatement;
+import org.apache.iotdb.db.mpp.plan.statement.metadata.CreateContinuousQueryStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.CreateFunctionStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.CreateMultiTimeSeriesStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.CreateTimeSeriesStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.CreateTriggerStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.DeleteStorageGroupStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.DeleteTimeSeriesStatement;
+import org.apache.iotdb.db.mpp.plan.statement.metadata.DropContinuousQueryStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.DropFunctionStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.DropTriggerStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.GetRegionIdStatement;
@@ -54,6 +56,7 @@ import org.apache.iotdb.db.mpp.plan.statement.metadata.ShowChildNodesStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.ShowChildPathsStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.ShowClusterStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.ShowConfigNodesStatement;
+import org.apache.iotdb.db.mpp.plan.statement.metadata.ShowContinuousQueriesStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.ShowDataNodesStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.ShowDevicesStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.ShowFunctionsStatement;
@@ -65,7 +68,6 @@ import org.apache.iotdb.db.mpp.plan.statement.metadata.ShowTriggersStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.UnSetTTLStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.template.ActivateTemplateStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.template.CreateSchemaTemplateStatement;
-import org.apache.iotdb.db.mpp.plan.statement.metadata.template.DeactivateTemplateStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.template.SetSchemaTemplateStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.template.ShowNodesInSchemaTemplateStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.template.ShowPathSetTemplateStatement;
@@ -415,4 +417,19 @@ public abstract class StatementVisitor<R, C> {
DeactivateTemplateStatement deactivateTemplateStatement, C context) {
return visitStatement(deactivateTemplateStatement, context);
}
+
+ public R visitCreateContinuousQuery(
+ CreateContinuousQueryStatement createContinuousQueryStatement, C context) {
+ return visitStatement(createContinuousQueryStatement, context);
+ }
+
+ public R visitDropContinuousQuery(
+ DropContinuousQueryStatement dropContinuousQueryStatement, C context) {
+ return visitStatement(dropContinuousQueryStatement, context);
+ }
+
+ public R visitShowContinuousQueries(
+ ShowContinuousQueriesStatement showContinuousQueriesStatement, C context) {
+ return visitStatement(showContinuousQueriesStatement, context);
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/statement/crud/QueryStatement.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/statement/crud/QueryStatement.java
index ae6586de5f..f05249ee6e 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/statement/crud/QueryStatement.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/statement/crud/QueryStatement.java
@@ -69,38 +69,40 @@ import java.util.Set;
*/
public class QueryStatement extends Statement {
- protected SelectComponent selectComponent;
- protected FromComponent fromComponent;
- protected WhereCondition whereCondition;
- protected HavingCondition havingCondition;
+ private SelectComponent selectComponent;
+ private FromComponent fromComponent;
+ private WhereCondition whereCondition;
+ private HavingCondition havingCondition;
// row limit and offset for result set. The default value is 0, which means no limit
- protected int rowLimit = 0;
+ private int rowLimit = 0;
// row offset for result set. The default value is 0
- protected int rowOffset = 0;
+ private int rowOffset = 0;
// series limit and offset for result set. The default value is 0, which means no limit
- protected int seriesLimit = 0;
+ private int seriesLimit = 0;
// series offset for result set. The default value is 0
- protected int seriesOffset = 0;
+ private int seriesOffset = 0;
- protected FillComponent fillComponent;
+ private FillComponent fillComponent;
- protected OrderByComponent orderByComponent;
+ private OrderByComponent orderByComponent;
- protected ResultSetFormat resultSetFormat = ResultSetFormat.ALIGN_BY_TIME;
+ private ResultSetFormat resultSetFormat = ResultSetFormat.ALIGN_BY_TIME;
// `GROUP BY TIME` clause
- protected GroupByTimeComponent groupByTimeComponent;
+ private GroupByTimeComponent groupByTimeComponent;
// `GROUP BY LEVEL` clause
- protected GroupByLevelComponent groupByLevelComponent;
+ private GroupByLevelComponent groupByLevelComponent;
// `GROUP BY TAG` clause
- protected GroupByTagComponent groupByTagComponent;
+ private GroupByTagComponent groupByTagComponent;
// `INTO` clause
- protected IntoComponent intoComponent;
+ private IntoComponent intoComponent;
+
+ private boolean isCqQueryBody;
public QueryStatement() {
this.statementType = StatementType.QUERY;
@@ -308,6 +310,14 @@ public class QueryStatement extends Statement {
return intoComponent != null;
}
+ public boolean isCqQueryBody() {
+ return isCqQueryBody;
+ }
+
+ public void setCqQueryBody(boolean cqQueryBody) {
+ isCqQueryBody = cqQueryBody;
+ }
+
public void semanticCheck() {
if (isAggregationQuery()) {
if (disableAlign()) {
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/statement/metadata/CreateContinuousQueryStatement.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/statement/metadata/CreateContinuousQueryStatement.java
new file mode 100644
index 0000000000..641a9365ba
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/statement/metadata/CreateContinuousQueryStatement.java
@@ -0,0 +1,192 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.mpp.plan.statement.metadata;
+
+import org.apache.iotdb.commons.cq.TimeoutPolicy;
+import org.apache.iotdb.commons.path.PartialPath;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.exception.sql.SemanticException;
+import org.apache.iotdb.db.mpp.plan.analyze.ExpressionAnalyzer;
+import org.apache.iotdb.db.mpp.plan.analyze.QueryType;
+import org.apache.iotdb.db.mpp.plan.constant.StatementType;
+import org.apache.iotdb.db.mpp.plan.statement.IConfigStatement;
+import org.apache.iotdb.db.mpp.plan.statement.Statement;
+import org.apache.iotdb.db.mpp.plan.statement.StatementVisitor;
+import org.apache.iotdb.db.mpp.plan.statement.component.GroupByTimeComponent;
+import org.apache.iotdb.db.mpp.plan.statement.crud.QueryStatement;
+
+import java.util.Collections;
+import java.util.List;
+
+public class CreateContinuousQueryStatement extends Statement implements IConfigStatement {
+
+ private String sql;
+ private String cqId;
+
+ // The query execution time interval, default value is group_by_interval in group by clause.
+ private long everyInterval;
+
+ // A date that represents the execution time of a certain cq task, default value is 0.
+ private long boundaryTime = 0;
+
+ // The start time of each query execution, default value is every_interval
+ private long startTimeOffset;
+
+ // The end time of each query execution, default value is 0.
+ private long endTimeOffset = 0;
+
+ // Specify how we deal with the cq task whose previous time interval execution is not finished
+ // while the next execution time has reached, default value is BLOCKED.
+ private TimeoutPolicy timeoutPolicy = TimeoutPolicy.BLOCKED;
+
+ private String queryBody;
+ private QueryStatement queryBodyStatement;
+
+ public CreateContinuousQueryStatement() {
+ super();
+ statementType = StatementType.CREATE_CONTINUOUS_QUERY;
+ }
+
+ public String getSql() {
+ return sql;
+ }
+
+ public void setSql(String sql) {
+ this.sql = sql;
+ }
+
+ public String getCqId() {
+ return cqId;
+ }
+
+ public void setCqId(String cqId) {
+ this.cqId = cqId;
+ }
+
+ public long getEveryInterval() {
+ return everyInterval;
+ }
+
+ public void setEveryInterval(long everyInterval) {
+ this.everyInterval = everyInterval;
+ }
+
+ public long getBoundaryTime() {
+ return boundaryTime;
+ }
+
+ public void setBoundaryTime(long boundaryTime) {
+ this.boundaryTime = boundaryTime;
+ }
+
+ public long getStartTimeOffset() {
+ return startTimeOffset;
+ }
+
+ public void setStartTimeOffset(long startTimeOffset) {
+ this.startTimeOffset = startTimeOffset;
+ }
+
+ public long getEndTimeOffset() {
+ return endTimeOffset;
+ }
+
+ public void setEndTimeOffset(long endTimeOffset) {
+ this.endTimeOffset = endTimeOffset;
+ }
+
+ public TimeoutPolicy getTimeoutPolicy() {
+ return timeoutPolicy;
+ }
+
+ public void setTimeoutPolicy(TimeoutPolicy timeoutPolicy) {
+ this.timeoutPolicy = timeoutPolicy;
+ }
+
+ public String getQueryBody() {
+ return queryBody;
+ }
+
+ public void setQueryBody(String queryBody) {
+ this.queryBody = queryBody;
+ }
+
+ public QueryStatement getQueryBodyStatement() {
+ return queryBodyStatement;
+ }
+
+ public void setQueryBodyStatement(QueryStatement queryBodyStatement) {
+ this.queryBodyStatement = queryBodyStatement;
+ }
+
+ public String getZoneId() {
+ return queryBodyStatement.getSelectComponent().getZoneId().getId();
+ }
+
+ @Override
+ public QueryType getQueryType() {
+ return QueryType.WRITE;
+ }
+
+ @Override
+ public List<? extends PartialPath> getPaths() {
+ return Collections.emptyList();
+ }
+
+ @Override
+ public <R, C> R accept(StatementVisitor<R, C> visitor, C context) {
+ return visitor.visitCreateContinuousQuery(this, context);
+ }
+
+ public void semanticCheck() {
+ if (everyInterval
+ < IoTDBDescriptor.getInstance().getConfig().getContinuousQueryMinimumEveryInterval()) {
+ throw new SemanticException(
+ "CQ: Every interval should not be lower than the `continuous_query_minimum_every_interval` configured.");
+ }
+ if (startTimeOffset <= 0) {
+ throw new SemanticException("CQ: The start time offset should be greater than 0.");
+ }
+ if (endTimeOffset < 0) {
+ throw new SemanticException("CQ: The end time offset should be greater than or equal to 0.");
+ }
+ if (startTimeOffset <= endTimeOffset) {
+ throw new SemanticException(
+ "CQ: The start time offset should be greater than end time offset.");
+ }
+ if (everyInterval > startTimeOffset) {
+ throw new SemanticException(
+ "CQ: The start time offset should be greater than or equal to every interval.");
+ }
+
+ if (!queryBodyStatement.isSelectInto()) {
+ throw new SemanticException("CQ: The query body is missing an INTO clause.");
+ }
+ GroupByTimeComponent groupByTimeComponent = queryBodyStatement.getGroupByTimeComponent();
+ if (groupByTimeComponent.getStartTime() != 0 || groupByTimeComponent.getEndTime() != 0) {
+ throw new SemanticException(
+ "CQ: Specifying time range in GROUP BY TIME clause is prohibited");
+ }
+ if (ExpressionAnalyzer.checkIfTimeFilterExist(
+ queryBodyStatement.getWhereCondition().getPredicate())) {
+ throw new SemanticException("CQ: Specifying time filters in the query body is prohibited.");
+ }
+ }
+}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/statement/metadata/DropContinuousQueryStatement.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/statement/metadata/DropContinuousQueryStatement.java
new file mode 100644
index 0000000000..85bc2382e6
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/statement/metadata/DropContinuousQueryStatement.java
@@ -0,0 +1,60 @@
+/*
+ * 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.mpp.plan.statement.metadata;
+
+import org.apache.iotdb.commons.path.PartialPath;
+import org.apache.iotdb.db.mpp.plan.analyze.QueryType;
+import org.apache.iotdb.db.mpp.plan.constant.StatementType;
+import org.apache.iotdb.db.mpp.plan.statement.IConfigStatement;
+import org.apache.iotdb.db.mpp.plan.statement.Statement;
+import org.apache.iotdb.db.mpp.plan.statement.StatementVisitor;
+
+import java.util.Collections;
+import java.util.List;
+
+public class DropContinuousQueryStatement extends Statement implements IConfigStatement {
+
+ private final String cqId;
+
+ public DropContinuousQueryStatement(String cqId) {
+ super();
+ statementType = StatementType.DROP_CONTINUOUS_QUERY;
+ this.cqId = cqId;
+ }
+
+ public String getCqId() {
+ return cqId;
+ }
+
+ @Override
+ public QueryType getQueryType() {
+ return QueryType.WRITE;
+ }
+
+ @Override
+ public List<? extends PartialPath> getPaths() {
+ return Collections.emptyList();
+ }
+
+ @Override
+ public <R, C> R accept(StatementVisitor<R, C> visitor, C context) {
+ return visitor.visitDropContinuousQuery(this, context);
+ }
+}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/statement/metadata/ShowContinuousQueriesStatement.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/statement/metadata/ShowContinuousQueriesStatement.java
new file mode 100644
index 0000000000..413a4c88de
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/statement/metadata/ShowContinuousQueriesStatement.java
@@ -0,0 +1,52 @@
+/*
+ * 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.mpp.plan.statement.metadata;
+
+import org.apache.iotdb.commons.path.PartialPath;
+import org.apache.iotdb.db.mpp.plan.analyze.QueryType;
+import org.apache.iotdb.db.mpp.plan.constant.StatementType;
+import org.apache.iotdb.db.mpp.plan.statement.IConfigStatement;
+import org.apache.iotdb.db.mpp.plan.statement.StatementVisitor;
+
+import java.util.Collections;
+import java.util.List;
+
+public class ShowContinuousQueriesStatement extends ShowStatement implements IConfigStatement {
+
+ public ShowContinuousQueriesStatement() {
+ super();
+ statementType = StatementType.SHOW_CONTINUOUS_QUERIES;
+ }
+
+ @Override
+ public QueryType getQueryType() {
+ return QueryType.READ;
+ }
+
+ @Override
+ public List<PartialPath> getPaths() {
+ return Collections.emptyList();
+ }
+
+ @Override
+ public <R, C> R accept(StatementVisitor<R, C> visitor, C context) {
+ return visitor.visitShowContinuousQueries(this, context);
+ }
+}
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 d6fa5100e1..7c3aafec55 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
@@ -151,7 +151,6 @@ import org.apache.iotdb.db.qp.logical.sys.UnSetTTLOperator;
import org.apache.iotdb.db.qp.logical.sys.UnloadFileOperator;
import org.apache.iotdb.db.qp.logical.sys.UnsetTemplateOperator;
import org.apache.iotdb.db.qp.sql.IoTDBSqlParser.ConstantContext;
-import org.apache.iotdb.db.qp.sql.IoTDBSqlParser.CqGroupByTimeClauseContext;
import org.apache.iotdb.db.qp.sql.IoTDBSqlParser.ExpressionContext;
import org.apache.iotdb.db.qp.utils.DateTimeUtils;
import org.apache.iotdb.db.query.executor.fill.IFill;
@@ -671,27 +670,30 @@ public class IoTDBSqlVisitor extends IoTDBSqlParserBaseVisitor<Operator> {
public Operator visitCreateContinuousQuery(IoTDBSqlParser.CreateContinuousQueryContext ctx) {
CreateContinuousQueryOperator createContinuousQueryOperator =
new CreateContinuousQueryOperator(SQLConstant.TOK_CONTINUOUS_QUERY_CREATE);
-
- createContinuousQueryOperator.setQuerySql(ctx.getText());
-
- createContinuousQueryOperator.setContinuousQueryName(
- parseIdentifier(ctx.continuousQueryName.getText()));
+ createContinuousQueryOperator.setContinuousQueryName(parseIdentifier(ctx.cqId.getText()));
if (ctx.resampleClause() != null) {
parseResampleClause(ctx.resampleClause(), createContinuousQueryOperator);
}
- parseCqSelectIntoClause(ctx.cqSelectIntoClause(), createContinuousQueryOperator);
+ if (ctx.timeoutPolicyClause() != null) {
+ throw new SQLParserException("CQ: CQ does not support to set timeout policy before v0.14.");
+ }
+
+ parseCqSelectIntoClause(ctx.selectStatement(), createContinuousQueryOperator);
StringBuilder sb = new StringBuilder();
sb.append("select ");
- sb.append(ctx.cqSelectIntoClause().selectClause().getText().substring(6));
+ sb.append(ctx.selectStatement().selectClause().getText().substring(6));
sb.append(" from ");
- sb.append(ctx.cqSelectIntoClause().fromClause().prefixPath(0).getText());
+ sb.append(ctx.selectStatement().fromClause().prefixPath(0).getText());
sb.append(" group by ([now() - ");
String groupByInterval =
- ctx.cqSelectIntoClause().cqGroupByTimeClause().DURATION_LITERAL().getText();
+ ((IoTDBSqlParser.GroupByTimeStatementContext) ctx.selectStatement().specialClause())
+ .groupByTimeClause()
+ .interval
+ .getText();
if (createContinuousQueryOperator.getForInterval() == 0) {
sb.append(groupByInterval);
} else {
@@ -731,7 +733,7 @@ public class IoTDBSqlVisitor extends IoTDBSqlParserBaseVisitor<Operator> {
}
public void parseCqSelectIntoClause(
- IoTDBSqlParser.CqSelectIntoClauseContext ctx,
+ IoTDBSqlParser.SelectStatementContext ctx,
CreateContinuousQueryOperator createContinuousQueryOperator) {
queryOp = new GroupByQueryOperator();
@@ -745,7 +747,7 @@ public class IoTDBSqlVisitor extends IoTDBSqlParserBaseVisitor<Operator> {
throw new SQLParserException("CQ: CQ currently does not support multiple series.");
}
- parseCqGroupByTimeClause(ctx.cqGroupByTimeClause(), createContinuousQueryOperator);
+ parseCqGroupByTimeClause(ctx.specialClause(), createContinuousQueryOperator);
if (queryOp.isGroupByLevel()) {
int[] groupByQueryLevels = queryOp.getSpecialClauseComponent().getLevels();
@@ -755,29 +757,36 @@ public class IoTDBSqlVisitor extends IoTDBSqlParserBaseVisitor<Operator> {
}
}
- createContinuousQueryOperator.setTargetPath(parseIntoPath(ctx.intoPath()));
+ createContinuousQueryOperator.setTargetPath(parseIntoPath(ctx.intoClause().intoPath(0)));
createContinuousQueryOperator.setQueryOperator(queryOp);
}
public void parseCqGroupByTimeClause(
- CqGroupByTimeClauseContext ctx, CreateContinuousQueryOperator createContinuousQueryOperator) {
+ IoTDBSqlParser.SpecialClauseContext ctx,
+ CreateContinuousQueryOperator createContinuousQueryOperator) {
+ if (!(ctx instanceof IoTDBSqlParser.GroupByTimeStatementContext)) {
+ throw new SQLParserException("CQ: Query body in CQ must be group by time query.");
+ }
+ IoTDBSqlParser.GroupByTimeClauseContext groupByCtx =
+ ((IoTDBSqlParser.GroupByTimeStatementContext) ctx).groupByTimeClause();
+
GroupByClauseComponent groupByClauseComponent = new GroupByClauseComponent();
long groupByInterval =
- parseTimeUnitOrSlidingStep(ctx.DURATION_LITERAL().getText(), true, groupByClauseComponent);
+ parseTimeUnitOrSlidingStep(groupByCtx.interval.getText(), true, groupByClauseComponent);
groupByClauseComponent.setUnit(groupByInterval);
createContinuousQueryOperator.setGroupByTimeInterval(groupByInterval);
- createContinuousQueryOperator.setGroupByTimeIntervalString(ctx.DURATION_LITERAL().getText());
+ createContinuousQueryOperator.setGroupByTimeIntervalString(groupByCtx.interval.getText());
groupByClauseComponent.setSlidingStep(groupByClauseComponent.getUnit());
groupByClauseComponent.setSlidingStepByMonth(groupByClauseComponent.isIntervalByMonth());
groupByClauseComponent.setLeftCRightO(true);
- if (ctx.LEVEL() != null && ctx.INTEGER_LITERAL() != null) {
- int[] levels = new int[ctx.INTEGER_LITERAL().size()];
- for (int i = 0; i < ctx.INTEGER_LITERAL().size(); i++) {
- levels[i] = Integer.parseInt(ctx.INTEGER_LITERAL().get(i).getText());
+ if (groupByCtx.LEVEL() != null && groupByCtx.INTEGER_LITERAL() != null) {
+ int[] levels = new int[groupByCtx.INTEGER_LITERAL().size()];
+ for (int i = 0; i < groupByCtx.INTEGER_LITERAL().size(); i++) {
+ levels[i] = Integer.parseInt(groupByCtx.INTEGER_LITERAL().get(i).getText());
}
groupByClauseComponent.setLevels(levels);
}
@@ -787,6 +796,10 @@ public class IoTDBSqlVisitor extends IoTDBSqlParserBaseVisitor<Operator> {
public void parseResampleClause(
IoTDBSqlParser.ResampleClauseContext ctx, CreateContinuousQueryOperator operator) {
+ if (ctx.RANGE() != null) {
+ throw new SQLParserException("CQ: CQ does not support to set RANGE before v0.14.");
+ }
+
if (ctx.DURATION_LITERAL().size() == 1) {
if (ctx.EVERY() != null) {
operator.setEveryInterval(
@@ -936,8 +949,7 @@ public class IoTDBSqlVisitor extends IoTDBSqlParserBaseVisitor<Operator> {
public Operator visitDropContinuousQuery(IoTDBSqlParser.DropContinuousQueryContext ctx) {
DropContinuousQueryOperator dropContinuousQueryOperator =
new DropContinuousQueryOperator(SQLConstant.TOK_CONTINUOUS_QUERY_DROP);
- dropContinuousQueryOperator.setContinuousQueryName(
- parseIdentifier(ctx.continuousQueryName.getText()));
+ dropContinuousQueryOperator.setContinuousQueryName(parseIdentifier(ctx.cqId.getText()));
return dropContinuousQueryOperator;
}