You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ro...@apache.org on 2022/01/14 14:20:42 UTC
[iotdb] branch iotdb-2353-cq updated: user doc and fix physical plan
This is an automated email from the ASF dual-hosted git repository.
rong pushed a commit to branch iotdb-2353-cq
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/iotdb-2353-cq by this push:
new 9182adf user doc and fix physical plan
9182adf is described below
commit 9182adfb816cb312920d7db9517c6dab85a3a071
Author: Steve Yurong Su <ro...@apache.org>
AuthorDate: Fri Jan 14 22:19:56 2022 +0800
user doc and fix physical plan
---
.../Advanced-Features/Continuous-Query.md | 97 ++++++++++++++----
.../Advanced-Features/Continuous-Query.md | 114 ++++++++++++++++-----
.../qp/physical/sys/CreateContinuousQueryPlan.java | 6 +-
3 files changed, 169 insertions(+), 48 deletions(-)
diff --git a/docs/UserGuide/Advanced-Features/Continuous-Query.md b/docs/UserGuide/Advanced-Features/Continuous-Query.md
index 505ee65..3ff2f43 100644
--- a/docs/UserGuide/Advanced-Features/Continuous-Query.md
+++ b/docs/UserGuide/Advanced-Features/Continuous-Query.md
@@ -21,8 +21,9 @@
# Continuous Query, CQ
-We can create, drop a CQ,
-and query all registered CQ configuration information through SQL statements.
+We can create, drop a CQ, and query all registered CQ configuration information through SQL statements.
+
+Note that the current distributed version of IoTDB does not support continuous queries. Please stay tuned.
## SQL statements
@@ -31,7 +32,7 @@ and query all registered CQ configuration information through SQL statements.
#### Syntax
```sql
-CREATE CONTINUOUS QUERY <cq_id>
+CREATE (CONTINUOUS QUERY | CQ) <cq_id>
[RESAMPLE EVERY <every_interval> FOR <for_interval>]
BEGIN
SELECT <function>(<path_suffix>) INTO <full_path> | <node_name>
@@ -43,6 +44,16 @@ END
* `<cq_id>` specifies the globally unique id of CQ.
* `<every_interval>` specifies the query execution time interval. We currently support the units of ns, us, ms, s, m, h, d, w, and its value should not be lower than the minimum threshold configured by the user.
* `<for_interval>` specifies the time range of each query as `[now()-<for_interval>, now())`. We currently support the units of ns, us, ms, s, m, h, d, w.
+* `<execution_boundary_time>` is a date that represents **the start time of the first window**.
+ * `<execution_boundary_time>` can be earlier than, equals to, later than **current time**.
+ * This parameter is optional. If not specified, it is equivalent to `BOUNDARY now()`.
+ * **The end time of the first window** is `<execution_boundary_time> + <for_interval>`.
+ * The **start time** of the `i (1 <= i)`th window is `<execution_boundary_time> + <for_interval> + (i - 1) * <every_interval>`.
+ * The **end time** of the `i (1 <= i)`th window is`<execution_boundary_time> + <for_interval> + i * <every_interval>`.
+ * If **current time** is earlier than or equal to **the end time of the first window**, then the first execution moment of the continuous query is **the end time of the first window**.
+ * If **current time** is later than **the end time of the first window**, then the first execution moment of the continuous query is the **end time of the first window whose end time is later than or equal to the current time** .
+ * The **query time range** at each execution moment is `[now() - <for_interval>, now())`.
+
* `<function>` specifies the aggregate function.
* `<path_prefix>` and `<path_suffix>` are spliced into the queried time series path.
* `<full_path>` or `<node_name>` specifies the result time series path.
@@ -106,8 +117,14 @@ It should be noted that the `x` in `${x}` should be greater than or equal to `1`
(If `<level>` is not specified, it should be less than or equal to the length of `<path_prefix>`). In the above example, `x` should be less than or equal to `2`.
##### Create `cq1`
-````
-CREATE CONTINUOUS QUERY cq1 BEGIN SELECT max_value(temperature) INTO temperature_max FROM root.ln.*.* GROUP BY time(10s) END
+````sql
+CREATE CONTINUOUS QUERY cq1
+BEGIN
+ SELECT max_value(temperature)
+ INTO temperature_max
+ FROM root.ln.*.*
+ GROUP BY time(10s)
+END
````
Query the maximum value of `root.ln.*.*.temperature` in the previous 10s every 10s (the results are grouped by 10s),
@@ -134,8 +151,15 @@ As a result, 4 new time series will be generated.
+-----------------------------+---------------------------------+---------------------------------+---------------------------------+---------------------------------+
````
##### Create `cq2`
-````
-CREATE CONTINUOUS QUERY cq2 RESAMPLE EVERY 20s FOR 20s BEGIN SELECT avg(temperature) INTO temperature_avg FROM root.ln.*.* GROUP BY time(10s), level=2 END
+````sql
+CREATE CONTINUOUS QUERY cq2
+RESAMPLE EVERY 20s FOR 20s
+BEGIN
+ SELECT avg(temperature)
+ INTO temperature_avg
+ FROM root.ln.*.*
+ GROUP BY time(10s), level=2
+END
````
Query the average value of `root.ln.*.*.temperature` in the previous 20s every 20s (the results are grouped by 10s),
and the results will be written to `root.${1}.${2}.temperature_avg`,
@@ -161,8 +185,15 @@ and `root.ln.wf01.temperature_avg` is generated by the aggregation calculation o
+-----------------------------+----------------------------+----------------------------+
````
##### Create `cq3`
-````
-CREATE CONTINUOUS QUERY cq3 RESAMPLE EVERY 20s FOR 20s BEGIN SELECT avg(temperature) INTO root.ln_cq.${2}.temperature_avg FROM root.ln.*.* GROUP BY time(10s), level=2 END
+````sql
+CREATE CONTINUOUS QUERY cq3
+RESAMPLE EVERY 20s FOR 20s
+BEGIN
+ SELECT avg(temperature)
+ INTO root.ln_cq.${2}.temperature_avg
+ FROM root.ln.*.*
+ GROUP BY time(10s), level=2
+END
````
The query mode is the same as `cq2`,
@@ -190,36 +221,62 @@ and `root.ln_cq.wf01.temperature_avg` is generated by the aggregation calculatio
+-----------------------------+-------------------------------+-------------------------------+
````
+##### 创建 `cq4`
+
+````sql
+CREATE CONTINUOUS QUERY cq4
+RESAMPLE EVERY 20s FOR 20s BOUNDARY 2022-01-14T23:00:00.000+08:00
+BEGIN
+ SELECT avg(temperature)
+ INTO root.ln_cq.${2}.temperature_avg
+ FROM root.ln.*.* GROUP BY time(10s), level=2
+END
+````
+
+This example is almost identical to creating cq3. The difference is that in this example the user specified `BOUNDARY 2022-01-14T23:00:00.000+08:00 `.
+
+Note that the first execution time of this CQ is later than the time in the example, so `2022-01-14T23:00:20.000+08:00` is the first execution time. Recursively, `2022-01-14T23:00:40.000+08:00` is the second execution moment, `2022-01-14T23:01:00.000+08:00` is the third execution moment...
+
+The SQL statement executed at the first execution moment is `select avg(temperature) from root.ln.*.* group by ([2022-01-14T23:00:00.000+08:00, 2022-01-14T23:00: 20.000+08:00), 10s), level = 2`.
+
+The SQL statement executed at the second execution moment is `select avg(temperature) from root.ln.*.* group by ([2022-01-14T23:00:20.000+08:00, 2022-01-14T23:00: 40.000+08:00), 10s), level = 2`.
+
+The SQL statement executed at the third execution moment is `select avg(temperature) from root.ln.*.* group by ([2022-01-14T23:00:40.000+08:00, 2022-01-14T23:01: 00.000+08:00), 10s), level = 2`.
+
+...
+
### Show CQ Information
+
#### Syntax
-````
+````sql
SHOW (CONTINUOUS QUERIES | CQS)
````
#### Example Result
````
-+-------+--------------+------------+----------------------------------------------------------------------------------------+-----------------------------------+
-|cq name|every interval|for interval| query sql| target path|
-+-------+--------------+------------+----------------------------------------------------------------------------------------+-----------------------------------+
-| cq1| 10000| 10000| select max_value(temperature) from root.ln.*.* group by ([now() - 10s, now()), 10s)|root.${1}.${2}.${3}.temperature_max|
-| cq3| 20000| 20000|select avg(temperature) from root.ln.*.* group by ([now() - 20s, now()), 10s), level = 2| root.ln_cq.${2}.temperature_avg|
-| cq2| 20000| 20000|select avg(temperature) from root.ln.*.* group by ([now() - 20s, now()), 10s), level = 2| root.${1}.${2}.temperature_avg|
-+-------+--------------+------------+----------------------------------------------------------------------------------------+-----------------------------------+
++-------+--------------+------------+-------------+----------------------------------------------------------------------------------------+-----------------------------------+
+|cq name|every interval|for interval| boundary| query sql| target path|
++-------+--------------+------------+-------------+----------------------------------------------------------------------------------------+-----------------------------------+
+| cq1| 10000| 10000|1642166102238| select max_value(temperature) from root.ln.*.* group by ([now() - 10s, now()), 10s)|root.${1}.${2}.${3}.temperature_max|
+| cq3| 20000| 20000|1642166118339|select avg(temperature) from root.ln.*.* group by ([now() - 20s, now()), 10s), level = 2| root.ln_cq.${2}.temperature_avg|
+| cq2| 20000| 20000|1642166111493|select avg(temperature) from root.ln.*.* group by ([now() - 20s, now()), 10s), level = 2| root.${1}.${2}.temperature_avg|
+| cq4| 20000| 20000|1642172400000|select avg(temperature) from root.ln.*.* group by ([now() - 20s, now()), 10s), level = 2| root.ln_cq.${2}.temperature_avg|
++-------+--------------+------------+-------------+----------------------------------------------------------------------------------------+-----------------------------------+
````
### Drop CQ
#### Syntax
-````
+````sql
DROP (CONTINUOUS QUERY | CQ) <cq_id>
````
#### Example
-````
+````sql
DROP CONTINUOUS QUERY cq3
````
-````
+````sql
DROP CQ cq3
````
## System Parameter Configuration
diff --git a/docs/zh/UserGuide/Advanced-Features/Continuous-Query.md b/docs/zh/UserGuide/Advanced-Features/Continuous-Query.md
index 73fa701..deda1ff 100644
--- a/docs/zh/UserGuide/Advanced-Features/Continuous-Query.md
+++ b/docs/zh/UserGuide/Advanced-Features/Continuous-Query.md
@@ -23,6 +23,8 @@
我们可以通过 SQL 语句注册、或卸载一个 CQ 实例，以及查询到所有已经注册的 CQ 配置信息。
+注意，目前连续查询尚未对分布式场景进行适配。敬请期待分布式版本。
+
## SQL 语句
### 创建 CQ
@@ -30,8 +32,8 @@
#### 语法
```sql
-CREATE CONTINUOUS QUERY <cq_id>
-[RESAMPLE EVERY <every_interval> FOR <for_interval>]
+CREATE (CONTINUOUS QUERY | CQ) <cq_id>
+[RESAMPLE EVERY <every_interval> FOR <for_interval> BOUNDARY <execution_boundary_time>]
BEGIN
SELECT <function>(<path_suffix>) INTO <full_path> | <node_name>
FROM <path_prefix>
@@ -43,18 +45,26 @@ END
* `<cq_id>` 指定 CQ 全局唯一的 id。
* `<every_interval>` 指定查询执行时间间隔，支持 ns、us、ms、s、m、h、d、w 等单位，其值不应小于用户所配置的 `continuous_query_min_every_interval` 值。可选择指定。
-* `<for_interval>` 指定每次查询的窗口大小，即查询时间范围为`[now() - <for_interval>, now())`，其中 `now()` 指查询时的时间戳。支持 ns、us、ms、s、m、h
- 、d、w 等单位。可选择指定。
+* `<for_interval>` 指定每次查询的窗口大小，即查询时间范围为`[now() - <for_interval>, now())`，其中 `now()` 指查询时的时间戳。支持 ns、us、ms、s、m、h、d、w 等单位。可选择指定。
+* `<execution_boundary_time>` 是一个日期参数，表示**第一个窗口的起始时间**。
+ * `<execution_boundary_time>` 可早于、等于、晚于**当前时间**。
+ * 该参数可选择指定，不指定的情况下等价于输入 `BOUNDARY now()`。
+ * **第一个窗口的结束时间**为 `<execution_boundary_time> + <for_interval>`。
+ * 第 ` i (1 <= i)` 个窗口的**开始时间** `<execution_boundary_time> + <for_interval> + （i - 1） * <every_interval>`。
+ * 第 ` i (1 <= i)` 个窗口的**结束时间** `<execution_boundary_time> + <for_interval> + i * <every_interval>`。
+ * 如果**当前时间**小于等于**第一个窗口的结束时间** ，那么连续查询的第一个执行时刻为**第一个窗口的结束时间**。
+ * 如果**当前时间**大于**第一个窗口的结束时间**，那么连续查询的第一个执行时刻为**第一个**大于等于**当前时间**的**窗口结束时间**。
+ * 每一个执行时刻执行的**查询时间范围**为`[now() - <for_interval>, now())`。
+
* `<function>` 指定聚合函数，目前支持 `count`, `sum`, `avg`, `last_value`, `first_value`, `min_time`, `max_time`, `min_value`, `max_value` 等。
* `<path_prefix>` 与 `<path_suffix>` 拼接成完整的查询原时间序列。
* `<full_path>` 或 `<node_name>` 指定将查询出的数据写入的结果序列路径。
-* `<group_by_interval>` 指定时间分组长度，支持 ns、us、ms、s、m、h
- 、d、w、mo、y 等单位。
+* `<group_by_interval>` 指定时间分组长度，支持 ns、us、ms、s、m、h、d、w、mo、y 等单位。
* `<level>`指按照序列第 `<level>` 层分组，将第 `<level>` 层同名的所有序列聚合。Group By Level 语句的具体语义及 `<level>` 的定义见 [路径层级分组聚合](../IoTDB-SQL-Language/DML-Data-Manipulation-Language.md)。
注：
-* `<for_interval>`,`<every_interval>` 可选择指定。如果用户没有指定其中的某一项，则未指定项的值按照`<group_by_interval>` 处理。
+* `<for_interval>`, `<every_interval>` 可选择指定。如果用户没有指定其中的某一项，则未指定项的值按照`<group_by_interval>` 处理。
* `<every_interval>`，`<for_interval>`，`<group_by_interval>` 的值均应大于 0。
* `<group_by_interval>` 的值应小于`<for_interval>`的值，否则系统会按照等于`<for_interval>`的值处理。
* 用户应当结合实际需求指定合适的 `<for_interval>` 与 `<every_interval>`。
@@ -108,13 +118,20 @@ END
（若未指定 `<level>`，则应小于等于 `<path_prefix>` 层级）。在上例中，`x` 应当小于等于 `2`。
##### 创建 `cq1`
-````
-CREATE CONTINUOUS QUERY cq1 BEGIN SELECT max_value(temperature) INTO temperature_max FROM root.ln.*.* GROUP BY time(10s) END
+````sql
+CREATE CONTINUOUS QUERY cq1
+BEGIN
+ SELECT max_value(temperature)
+ INTO temperature_max
+ FROM root.ln.*.*
+ GROUP BY time(10s)
+END
````
每隔 10s 查询 `root.ln.*.*.temperature` 在前 10s 内的最大值（结果以 10s 为一组），
将结果写入到 `root.${1}.${2}.${3}.temperature_max` 中，
结果将产生 4 条新序列：
+
````
+---------------------------------+-----+-------------+--------+--------+-----------+----+----------+
| timeseries|alias|storage group|dataType|encoding|compression|tags|attributes|
@@ -136,8 +153,15 @@ CREATE CONTINUOUS QUERY cq1 BEGIN SELECT max_value(temperature) INTO temperature
+-----------------------------+---------------------------------+---------------------------------+---------------------------------+---------------------------------+
````
##### 创建 `cq2`
-````
-CREATE CONTINUOUS QUERY cq2 RESAMPLE EVERY 20s FOR 20s BEGIN SELECT avg(temperature) INTO temperature_avg FROM root.ln.*.* GROUP BY time(10s), level=2 END
+````sql
+CREATE CONTINUOUS QUERY cq2
+RESAMPLE EVERY 20s FOR 20s
+BEGIN
+ SELECT avg(temperature)
+ INTO temperature_avg
+ FROM root.ln.*.*
+ GROUP BY time(10s), level=2
+END
````
每隔 20s 查询 `root.ln.*.*.temperature` 在前 20s 内的平均值（结果以 10s 为一组，按照第 2 层节点分组），
@@ -165,13 +189,21 @@ CREATE CONTINUOUS QUERY cq2 RESAMPLE EVERY 20s FOR 20s BEGIN SELECT avg(temperat
+-----------------------------+----------------------------+----------------------------+
````
##### 创建 `cq3`
-````
-CREATE CONTINUOUS QUERY cq3 RESAMPLE EVERY 20s FOR 20s BEGIN SELECT avg(temperature) INTO root.ln_cq.${2}.temperature_avg FROM root.ln.*.* GROUP BY time(10s), level=2 END
+````sql
+CREATE CONTINUOUS QUERY cq3
+RESAMPLE EVERY 20s FOR 20s
+BEGIN
+ SELECT avg(temperature)
+ INTO root.ln_cq.${2}.temperature_avg
+ FROM root.ln.*.*
+ GROUP BY time(10s), level=2
+END
````
查询模式与 cq2 相同，在这个例子中，用户自行指定结果写入到 `root.ln_cq.${2}.temperature_avg` 中。
结果将产生如下两条新序列，
其中 `root.ln_cq.wf02.temperature_avg` 由 `root.ln.wf02.wt02.temperature` 和 `root.ln.wf02.wt01.temperature` 聚合计算生成，
`root.ln_cq.wf01.temperature_avg` 由 `root.ln.wf01.wt02.temperature` 和 `root.ln.wf01.wt01.temperature` 聚合计算生成。
+
````
+-------------------------------+-----+-------------+--------+--------+-----------+----+----------+
| timeseries|alias|storage group|dataType|encoding|compression|tags|attributes|
@@ -191,32 +223,64 @@ CREATE CONTINUOUS QUERY cq3 RESAMPLE EVERY 20s FOR 20s BEGIN SELECT avg(temperat
+-----------------------------+-------------------------------+-------------------------------+
````
+##### 创建 `cq4`
+
+````sql
+CREATE CONTINUOUS QUERY cq4
+RESAMPLE EVERY 20s FOR 20s BOUNDARY 2022-01-14T23:00:00.000+08:00
+BEGIN
+ SELECT avg(temperature)
+ INTO root.ln_cq.${2}.temperature_avg
+ FROM root.ln.*.* GROUP BY time(10s), level=2
+END
+````
+
+这个例子与创建 cq3 几乎完全相同。不同的是，在这个例子中用户自行指定了 `BOUNDARY 2022-01-14T23:00:00.000+08:00 ` 。
+
+注意这个 CQ 的第一个执行时刻大于例子中的时间，因此 `2022-01-14T23:00:20.000+08:00` 为第一个执行时刻。递推地，`2022-01-14T23:00:40.000+08:00` 为第二个执行时刻，`2022-01-14T23:01:00.000+08:00` 为第三个执行时刻……
+
+第一个执行时刻执行的 SQL 语句为 `select avg(temperature) from root.ln.*.* group by ([2022-01-14T23:00:00.000+08:00, 2022-01-14T23:00:20.000+08:00), 10s), level = 2`。
+
+第二个执行时刻执行的 SQL 语句为 `select avg(temperature) from root.ln.*.* group by ([2022-01-14T23:00:20.000+08:00, 2022-01-14T23:00:40.000+08:00), 10s), level = 2`。
+
+第三个执行时刻执行的 SQL 语句为 `select avg(temperature) from root.ln.*.* group by ([2022-01-14T23:00:40.000+08:00, 2022-01-14T23:01:00.000+08:00), 10s), level = 2`。
+
+……
+
+
### 展示 CQ 信息
+
#### 语法
-````
-SHOW CONTINUOUS QUERIES
+````sql
+SHOW (CONTINUOUS QUERIES | CQS)
````
#### 结果示例
````
-+-------+--------------+------------+----------------------------------------------------------------------------------------+-----------------------------------+
-|cq name|every interval|for interval| query sql| target path|
-+-------+--------------+------------+----------------------------------------------------------------------------------------+-----------------------------------+
-| cq1| 10000| 10000| select max_value(temperature) from root.ln.*.* group by ([now() - 10s, now()), 10s)|root.${1}.${2}.${3}.temperature_max|
-| cq3| 20000| 20000|select avg(temperature) from root.ln.*.* group by ([now() - 20s, now()), 10s), level = 2| root.ln_cq.${2}.temperature_avg|
-| cq2| 20000| 20000|select avg(temperature) from root.ln.*.* group by ([now() - 20s, now()), 10s), level = 2| root.${1}.${2}.temperature_avg|
-+-------+--------------+------------+----------------------------------------------------------------------------------------+-----------------------------------+
++-------+--------------+------------+-------------+----------------------------------------------------------------------------------------+-----------------------------------+
+|cq name|every interval|for interval| boundary| query sql| target path|
++-------+--------------+------------+-------------+----------------------------------------------------------------------------------------+-----------------------------------+
+| cq1| 10000| 10000|1642166102238| select max_value(temperature) from root.ln.*.* group by ([now() - 10s, now()), 10s)|root.${1}.${2}.${3}.temperature_max|
+| cq3| 20000| 20000|1642166118339|select avg(temperature) from root.ln.*.* group by ([now() - 20s, now()), 10s), level = 2| root.ln_cq.${2}.temperature_avg|
+| cq2| 20000| 20000|1642166111493|select avg(temperature) from root.ln.*.* group by ([now() - 20s, now()), 10s), level = 2| root.${1}.${2}.temperature_avg|
+| cq4| 20000| 20000|1642172400000|select avg(temperature) from root.ln.*.* group by ([now() - 20s, now()), 10s), level = 2| root.ln_cq.${2}.temperature_avg|
++-------+--------------+------------+-------------+----------------------------------------------------------------------------------------+-----------------------------------+
````
### 删除 CQ
#### 语法
-````
-DROP CONTINUOUS QUERY <cq_id>
+````sql
+DROP (CONTINUOUS QUERY | CQ) <cq_id>
````
#### 示例
-````
+````sql
DROP CONTINUOUS QUERY cq3
````
+``` sql
+DROP CQ cq3
+```
+
## 系统参数配置
+
| 参数名 | 描述 | 数据类型| 默认值 |
| :---------------------------------- |-------- | ----| -----|
| `continuous_query_execution_thread` | 执行连续查询任务的线程池的线程数 | int | max(1, CPU 核数 / 2)|
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/CreateContinuousQueryPlan.java b/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/CreateContinuousQueryPlan.java
index eed9fb7..79ad566 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/CreateContinuousQueryPlan.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/CreateContinuousQueryPlan.java
@@ -57,12 +57,12 @@ public class CreateContinuousQueryPlan extends PhysicalPlan {
String groupByTimeIntervalString,
Long firstExecutionTimeBoundary) {
super(Operator.OperatorType.CREATE_CONTINUOUS_QUERY);
- querySql = querySql.toLowerCase();
this.querySql = querySql;
- int indexOfGroupBy = querySql.indexOf("group by");
+ String querySqlLowerCase = querySql.toLowerCase();
+ int indexOfGroupBy = querySqlLowerCase.indexOf("group by");
this.querySqlBeforeGroupByClause =
indexOfGroupBy == -1 ? querySql : querySql.substring(0, indexOfGroupBy);
- int indexOfLevel = querySql.indexOf("level");
+ int indexOfLevel = querySqlLowerCase.indexOf("level");
this.querySqlAfterGroupByClause = indexOfLevel == -1 ? "" : querySql.substring(indexOfLevel);
this.continuousQueryName = continuousQueryName;
this.targetPath = targetPath;