You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by hu...@apache.org on 2022/10/25 07:04:15 UTC
[iotdb] branch lmh/mppCQ updated (6a94ae4bba -> f53711e553)
This is an automated email from the ASF dual-hosted git repository.
hui pushed a change to branch lmh/mppCQ
in repository https://gitbox.apache.org/repos/asf/iotdb.git
from 6a94ae4bba compatible with old version
new 274e80518a add cq related statements
new 7299a8f39e finish sql parser
new 15f2f82f3b add cq related tasks
new f53711e553 Implement basic task executor
The 4 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../org/apache/iotdb/db/qp/sql/IoTDBSqlParser.g4 | 2 +-
.../db/mpp/common/header/ColumnHeaderConstant.java | 11 ++
.../db/mpp/common/header/DatasetHeaderFactory.java | 4 +
.../plan/execution/config/ConfigTaskVisitor.java | 24 ++++
.../config/executor/ClusterConfigTaskExecutor.java | 77 +++++++++++
.../config/executor/IConfigTaskExecutor.java | 8 ++
.../executor/StandaloneConfigTaskExecutor.java | 46 +++++++
...ionTask.java => CreateContinuousQueryTask.java} | 12 +-
...ctionTask.java => DropContinuousQueryTask.java} | 12 +-
...TTLTask.java => ShowContinuousQueriesTask.java} | 43 +++---
.../iotdb/db/mpp/plan/parser/ASTVisitor.java | 108 +++++++++++++--
.../db/mpp/plan/statement/StatementVisitor.java | 18 +++
.../metadata/CreateContinuousQueryStatement.java | 153 +++++++++++++++++++++
...ment.java => DropContinuousQueryStatement.java} | 24 ++--
...nt.java => ShowContinuousQueriesStatement.java} | 16 +--
15 files changed, 489 insertions(+), 69 deletions(-)
copy server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/metadata/{DropFunctionTask.java => CreateContinuousQueryTask.java} (72%)
copy server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/metadata/{DropFunctionTask.java => DropContinuousQueryTask.java} (77%)
copy server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/metadata/{ShowTTLTask.java => ShowContinuousQueriesTask.java} (64%)
create mode 100644 server/src/main/java/org/apache/iotdb/db/mpp/plan/statement/metadata/CreateContinuousQueryStatement.java
copy server/src/main/java/org/apache/iotdb/db/mpp/plan/statement/metadata/{DropFunctionStatement.java => DropContinuousQueryStatement.java} (81%)
copy server/src/main/java/org/apache/iotdb/db/mpp/plan/statement/metadata/{ShowTriggersStatement.java => ShowContinuousQueriesStatement.java} (85%)
[iotdb] 03/04: add cq related tasks
Posted by hu...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hui pushed a commit to branch lmh/mppCQ
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 15f2f82f3bdc3ebc8ab3d4e089e1028a21ed3cc8
Author: Minghui Liu <li...@foxmail.com>
AuthorDate: Tue Oct 25 11:10:48 2022 +0800
add cq related tasks
---
.../plan/execution/config/ConfigTaskVisitor.java | 24 +++++++++++++
.../config/metadata/CreateContinuousQueryTask.java | 42 ++++++++++++++++++++++
.../config/metadata/DropContinuousQueryTask.java} | 30 +++++++---------
.../metadata/ShowContinuousQueriesTask.java} | 29 +++++----------
.../db/mpp/plan/statement/StatementVisitor.java | 18 ++++++++++
.../metadata/CreateContinuousQueryStatement.java | 6 ++++
.../metadata/DropContinuousQueryStatement.java | 6 ++++
.../metadata/ShowContinuousQueriesStatement.java | 6 ++++
8 files changed, 124 insertions(+), 37 deletions(-)
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 7fc68c0d46..92c29a5f4a 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;
@@ -63,10 +66,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;
@@ -76,6 +81,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;
@@ -344,6 +350,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/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..c7e8617e44
--- /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 null;
+ }
+}
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/execution/config/metadata/DropContinuousQueryTask.java
similarity index 50%
copy from server/src/main/java/org/apache/iotdb/db/mpp/plan/statement/metadata/ShowContinuousQueriesStatement.java
copy to server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/metadata/DropContinuousQueryTask.java
index 94edcdd8d5..a0c5a953f5 100644
--- 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/execution/config/metadata/DropContinuousQueryTask.java
@@ -17,30 +17,26 @@
* under the License.
*/
-package org.apache.iotdb.db.mpp.plan.statement.metadata;
+package org.apache.iotdb.db.mpp.plan.execution.config.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.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 java.util.Collections;
-import java.util.List;
+import com.google.common.util.concurrent.ListenableFuture;
-public class ShowContinuousQueriesStatement extends ShowStatement implements IConfigStatement {
+public class DropContinuousQueryTask implements IConfigTask {
- public ShowContinuousQueriesStatement() {
- super();
- statementType = StatementType.SHOW_CONTINUOUS_QUERIES;
- }
+ private final String cqId;
- @Override
- public QueryType getQueryType() {
- return QueryType.READ;
+ public DropContinuousQueryTask(DropContinuousQueryStatement dropContinuousQueryStatement) {
+ this.cqId = dropContinuousQueryStatement.getCqId();
}
@Override
- public List<PartialPath> getPaths() {
- return Collections.emptyList();
+ public ListenableFuture<ConfigTaskResult> execute(IConfigTaskExecutor configTaskExecutor)
+ throws InterruptedException {
+ return null;
}
}
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/execution/config/metadata/ShowContinuousQueriesTask.java
similarity index 53%
copy from server/src/main/java/org/apache/iotdb/db/mpp/plan/statement/metadata/ShowContinuousQueriesStatement.java
copy to server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/metadata/ShowContinuousQueriesTask.java
index 94edcdd8d5..249f0dfd3a 100644
--- 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/execution/config/metadata/ShowContinuousQueriesTask.java
@@ -17,30 +17,19 @@
* under the License.
*/
-package org.apache.iotdb.db.mpp.plan.statement.metadata;
+package org.apache.iotdb.db.mpp.plan.execution.config.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.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 java.util.Collections;
-import java.util.List;
+import com.google.common.util.concurrent.ListenableFuture;
-public class ShowContinuousQueriesStatement extends ShowStatement implements IConfigStatement {
-
- public ShowContinuousQueriesStatement() {
- super();
- statementType = StatementType.SHOW_CONTINUOUS_QUERIES;
- }
-
- @Override
- public QueryType getQueryType() {
- return QueryType.READ;
- }
+public class ShowContinuousQueriesTask implements IConfigTask {
@Override
- public List<PartialPath> getPaths() {
- return Collections.emptyList();
+ public ListenableFuture<ConfigTaskResult> execute(IConfigTaskExecutor configTaskExecutor)
+ throws InterruptedException {
+ return null;
}
}
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 be99227d27..2559f09c4f 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;
@@ -409,4 +412,19 @@ public abstract class StatementVisitor<R, C> {
public R visitGetTimeSlotList(GetTimeSlotListStatement getTimeSlotListStatement, C context) {
return visitStatement(getTimeSlotListStatement, 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/metadata/CreateContinuousQueryStatement.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/statement/metadata/CreateContinuousQueryStatement.java
index fd6120b6da..9ff67cc071 100644
--- 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
@@ -25,6 +25,7 @@ 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.crud.QueryStatement;
import java.util.Collections;
@@ -131,4 +132,9 @@ public class CreateContinuousQueryStatement extends Statement implements IConfig
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);
+ }
}
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
index fe35672421..85bc2382e6 100644
--- 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
@@ -24,6 +24,7 @@ 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;
@@ -51,4 +52,9 @@ public class DropContinuousQueryStatement extends Statement implements IConfigSt
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
index 94edcdd8d5..413a4c88de 100644
--- 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
@@ -23,6 +23,7 @@ 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;
@@ -43,4 +44,9 @@ public class ShowContinuousQueriesStatement extends ShowStatement implements ICo
public List<PartialPath> getPaths() {
return Collections.emptyList();
}
+
+ @Override
+ public <R, C> R accept(StatementVisitor<R, C> visitor, C context) {
+ return visitor.visitShowContinuousQueries(this, context);
+ }
}
[iotdb] 04/04: Implement basic task executor
Posted by hu...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hui pushed a commit to branch lmh/mppCQ
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit f53711e5531f8f3ce41830fa38c7ca60c9fdbb15
Author: Minghui Liu <li...@foxmail.com>
AuthorDate: Tue Oct 25 15:03:40 2022 +0800
Implement basic task executor
---
.../db/mpp/common/header/ColumnHeaderConstant.java | 11 ++++
.../db/mpp/common/header/DatasetHeaderFactory.java | 4 ++
.../config/executor/ClusterConfigTaskExecutor.java | 77 ++++++++++++++++++++++
.../config/executor/IConfigTaskExecutor.java | 8 +++
.../executor/StandaloneConfigTaskExecutor.java | 46 +++++++++++++
.../config/metadata/CreateContinuousQueryTask.java | 2 +-
.../config/metadata/DropContinuousQueryTask.java | 2 +-
.../config/metadata/ShowContinuousQueriesTask.java | 42 +++++++++++-
.../iotdb/db/mpp/plan/parser/ASTVisitor.java | 1 +
.../metadata/CreateContinuousQueryStatement.java | 13 ++++
10 files changed, 203 insertions(+), 3 deletions(-)
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/execution/config/executor/ClusterConfigTaskExecutor.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/executor/ClusterConfigTaskExecutor.java
index 4857dbcd40..3e7e264756 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,10 +32,12 @@ 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.TCreateTriggerReq;
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;
@@ -52,6 +54,7 @@ import org.apache.iotdb.confignode.rpc.thrift.TGetTriggerTableResp;
import org.apache.iotdb.confignode.rpc.thrift.TPipeInfo;
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;
@@ -76,6 +79,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;
@@ -87,6 +91,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;
@@ -1115,4 +1120,76 @@ public class ClusterConfigTaskExecutor implements IConfigTaskExecutor {
GetTimeSlotListTask.buildTSBlock(resp, future);
return future;
}
+
+ @Override
+ public SettableFuture<ConfigTaskResult> createContinuousQuery(
+ CreateContinuousQueryStatement createContinuousQueryStatement) {
+ 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 9cbae3e4f6..b8eaa69050 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;
@@ -141,4 +142,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 a934b815c7..af40b74604 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;
@@ -620,4 +621,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
index c7e8617e44..060162e731 100644
--- 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
@@ -37,6 +37,6 @@ public class CreateContinuousQueryTask implements IConfigTask {
@Override
public ListenableFuture<ConfigTaskResult> execute(IConfigTaskExecutor configTaskExecutor)
throws InterruptedException {
- return null;
+ 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
index a0c5a953f5..17a76a56fa 100644
--- 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
@@ -37,6 +37,6 @@ public class DropContinuousQueryTask implements IConfigTask {
@Override
public ListenableFuture<ConfigTaskResult> execute(IConfigTaskExecutor configTaskExecutor)
throws InterruptedException {
- return null;
+ 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
index 249f0dfd3a..deb4fb10e6 100644
--- 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
@@ -19,17 +19,57 @@
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 null;
+ 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 2ba49c7ed7..767c37fccb 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
@@ -814,6 +814,7 @@ public class ASTVisitor extends IoTDBSqlParserBaseVisitor<Statement> {
@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());
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
index 9ff67cc071..7857fd3685 100644
--- 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
@@ -33,6 +33,7 @@ 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.
@@ -59,6 +60,14 @@ public class CreateContinuousQueryStatement extends Statement implements IConfig
statementType = StatementType.CREATE_CONTINUOUS_QUERY;
}
+ public String getSql() {
+ return sql;
+ }
+
+ public void setSql(String sql) {
+ this.sql = sql;
+ }
+
public String getCqId() {
return cqId;
}
@@ -123,6 +132,10 @@ public class CreateContinuousQueryStatement extends Statement implements IConfig
this.queryBodyStatement = queryBodyStatement;
}
+ public String getZoneId() {
+ return queryBodyStatement.getSelectComponent().getZoneId().getId();
+ }
+
@Override
public QueryType getQueryType() {
return QueryType.WRITE;
[iotdb] 01/04: add cq related statements
Posted by hu...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hui pushed a commit to branch lmh/mppCQ
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 274e80518ae8d0803445f58e70fd86a2b9736120
Author: Minghui Liu <li...@foxmail.com>
AuthorDate: Tue Oct 25 10:10:55 2022 +0800
add cq related statements
---
.../metadata/CreateContinuousQueryStatement.java | 121 +++++++++++++++++++++
.../metadata/DropContinuousQueryStatement.java | 54 +++++++++
.../metadata/ShowContinuousQueriesStatement.java | 46 ++++++++
3 files changed, 221 insertions(+)
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..01589fc5d8
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/statement/metadata/CreateContinuousQueryStatement.java
@@ -0,0 +1,121 @@
+/*
+ * 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.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 java.util.Collections;
+import java.util.List;
+
+public class CreateContinuousQueryStatement extends Statement implements IConfigStatement {
+
+ private String cqId;
+ private long everyInterval;
+ private long boundaryTime;
+ private long startTimeOffset;
+ private long endTimeOffset;
+ private TimeoutPolicy timeoutPolicy;
+ private String queryBody;
+ private String sql;
+
+ public CreateContinuousQueryStatement() {
+ super();
+ statementType = StatementType.CREATE_CONTINUOUS_QUERY;
+ }
+
+ 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 String getSql() {
+ return sql;
+ }
+
+ public void setSql(String sql) {
+ this.sql = sql;
+ }
+
+ @Override
+ public QueryType getQueryType() {
+ return QueryType.WRITE;
+ }
+
+ @Override
+ public List<? extends PartialPath> getPaths() {
+ return Collections.emptyList();
+ }
+}
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..fe35672421
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/statement/metadata/DropContinuousQueryStatement.java
@@ -0,0 +1,54 @@
+/*
+ * 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 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();
+ }
+}
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..94edcdd8d5
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/statement/metadata/ShowContinuousQueriesStatement.java
@@ -0,0 +1,46 @@
+/*
+ * 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 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();
+ }
+}
[iotdb] 02/04: finish sql parser
Posted by hu...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hui pushed a commit to branch lmh/mppCQ
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 7299a8f39ed158332dbdd10c27617cbdededbab6
Author: Minghui Liu <li...@foxmail.com>
AuthorDate: Tue Oct 25 10:51:37 2022 +0800
finish sql parser
---
.../org/apache/iotdb/db/qp/sql/IoTDBSqlParser.g4 | 2 +-
.../iotdb/db/mpp/plan/parser/ASTVisitor.java | 107 ++++++++++++++++++---
.../metadata/CreateContinuousQueryStatement.java | 29 ++++--
3 files changed, 115 insertions(+), 23 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 5710bbbb57..7e51b11748 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
@@ -164,7 +164,7 @@ resampleClause
: RESAMPLE
(EVERY everyInterval=DURATION_LITERAL)?
(FOR DURATION_LITERAL)?
- (BOUNDARY executionBoundaryTime=dateExpression)?
+ (BOUNDARY boundaryTime=dateExpression)?
(RANGE startTimeOffset=DURATION_LITERAL (COMMA endTimeOffset=DURATION_LITERAL))?
;
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 d599436580..2ba49c7ed7 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;
@@ -774,11 +778,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();
@@ -804,6 +810,79 @@ public class ASTVisitor extends IoTDBSqlParserBaseVisitor<Statement> {
}
}
+ // Create CQ =====================================================================
+ @Override
+ public Statement visitCreateContinuousQuery(IoTDBSqlParser.CreateContinuousQueryContext ctx) {
+ CreateContinuousQueryStatement statement = new CreateContinuousQueryStatement();
+ statement.setCqId(parseIdentifier(ctx.cqId.getText()));
+
+ statement.setQueryBody(ctx.selectStatement().getText());
+ statement.setQueryBodyStatement((QueryStatement) visitSelectStatement(ctx.selectStatement()));
+
+ 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 ========================================================================
@@ -951,23 +1030,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());
@@ -1005,23 +1084,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/metadata/CreateContinuousQueryStatement.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/statement/metadata/CreateContinuousQueryStatement.java
index 01589fc5d8..fd6120b6da 100644
--- 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
@@ -25,6 +25,7 @@ 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.crud.QueryStatement;
import java.util.Collections;
import java.util.List;
@@ -32,13 +33,25 @@ import java.util.List;
public class CreateContinuousQueryStatement extends Statement implements IConfigStatement {
private String cqId;
+
+ // The query execution time interval, default value is group_by_interval in group by clause.
private long everyInterval;
- private long boundaryTime;
+
+ // 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;
- private long endTimeOffset;
- private TimeoutPolicy timeoutPolicy;
+
+ // 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 String sql;
+ private QueryStatement queryBodyStatement;
public CreateContinuousQueryStatement() {
super();
@@ -101,12 +114,12 @@ public class CreateContinuousQueryStatement extends Statement implements IConfig
this.queryBody = queryBody;
}
- public String getSql() {
- return sql;
+ public QueryStatement getQueryBodyStatement() {
+ return queryBodyStatement;
}
- public void setSql(String sql) {
- this.sql = sql;
+ public void setQueryBodyStatement(QueryStatement queryBodyStatement) {
+ this.queryBodyStatement = queryBodyStatement;
}
@Override