You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by qi...@apache.org on 2020/06/06 11:29:04 UTC
[incubator-iotdb] 01/02: [IOTDB-622] add count records for all
databases
This is an automated email from the ASF dual-hosted git repository.
qiaojialin pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-iotdb.git
commit 27c2ce015b84a17d9cb3a5539c79c48db3f1d29a
Author: mychaow <94...@qq.com>
AuthorDate: Wed May 13 11:26:20 2020 +0800
[IOTDB-622] add count records for all databases
---
docs/SystemDesign/DataQuery/AlignByDeviceQuery.md | 4 +-
docs/SystemDesign/DataQuery/GroupByQuery.md | 25 ++-
.../DML Data Manipulation Language.md | 63 +++++++
.../SystemDesign/DataQuery/AlignByDeviceQuery.md | 4 +-
docs/zh/SystemDesign/DataQuery/GroupByQuery.md | 24 ++-
.../DML Data Manipulation Language.md | 64 +++++++
.../iotdb/flink/tsfile/RowRowRecordParser.java | 2 +-
server/pom.xml | 10 +
.../org/apache/iotdb/db/qp/strategy/SqlBase.g4 | 11 +-
.../org/apache/iotdb/db/auth/AuthorityChecker.java | 2 +-
.../db/conf/adapter/IoTDBConfigDynamicAdapter.java | 2 +-
.../apache/iotdb/db/qp/executor/PlanExecutor.java | 24 +--
.../org/apache/iotdb/db/qp/logical/Operator.java | 2 +-
.../iotdb/db/qp/logical/crud/QueryOperator.java | 27 ++-
.../db/qp/physical/crud/AlignByDevicePlan.java | 12 +-
...oupByFillPlan.java => GroupByFillTimePlan.java} | 4 +-
.../{GroupByPlan.java => GroupByTimePlan.java} | 16 +-
.../iotdb/db/qp/physical/crud/QueryPlan.java | 10 +
.../iotdb/db/qp/strategy/LogicalGenerator.java | 41 ++--
.../iotdb/db/qp/strategy/PhysicalGenerator.java | 72 +++----
.../db/query/dataset/AlignByDeviceDataSet.java | 24 +--
.../dataset/groupby/GroupByEngineDataSet.java | 16 +-
.../query/dataset/groupby/GroupByFillDataSet.java | 12 +-
.../query/dataset/groupby/GroupByLevelDataSet.java | 208 +++++++++++++++++++++
.../groupby/GroupByWithValueFilterDataSet.java | 26 +--
.../groupby/GroupByWithoutValueFilterDataSet.java | 16 +-
.../iotdb/db/query/executor/IQueryRouter.java | 4 +-
.../iotdb/db/query/executor/QueryRouter.java | 75 +++++---
.../org/apache/iotdb/db/service/TSServiceImpl.java | 29 +--
.../org/apache/iotdb/db/utils/FilePathUtils.java | 44 +++++
.../java/org/apache/iotdb/db/qp/PlannerTest.java | 2 +-
.../apache/iotdb/db/qp/plan/PhysicalPlanTest.java | 18 +-
.../db/query/dataset/GroupByLevelDataSetTest.java | 189 +++++++++++++++++++
.../query/executor/GroupByEngineDataSetTest.java | 62 +++---
.../apache/iotdb/tsfile/read/common/RowRecord.java | 8 +
35 files changed, 926 insertions(+), 226 deletions(-)
diff --git a/docs/SystemDesign/DataQuery/AlignByDeviceQuery.md b/docs/SystemDesign/DataQuery/AlignByDeviceQuery.md
index ca37948..6df28dc 100644
--- a/docs/SystemDesign/DataQuery/AlignByDeviceQuery.md
+++ b/docs/SystemDesign/DataQuery/AlignByDeviceQuery.md
@@ -40,7 +40,7 @@ First explain the meaning of some important fields in AlignByDevicePlan:
- `Map<String, TSDataType> measurementDataTypeMap`：AlignByDevicePlan requires that the data type of the sensor of the same name be the same for different devices. This field is a Map structure of `measurementName-> dataType`. For example `root.sg.d1.s1` and` root.sg.d2.s1` should be of the same data type.
- `enum MeasurementType`：Three measurement types are recorded. Measurements that do not exist in any device are of type `NonExist`; measurements with single or double quotes are of type` Constant`; measurements that exist are of type `Exist`.
- `Map<String, MeasurementType> measurementTypeMap`: This field is a Map structure of `measureName-> measurementType`, which is used to record all measurement types in the query.
-- groupByPlan, fillQueryPlan, aggregationPlan：To avoid redundancy, these three execution plans are set as subclasses of RawDataQueryPlan and set as variables in AlignByDevicePlan. If the query plan belongs to one of these three plans, the field is assigned and saved.
+- groupByTimePlan, fillQueryPlan, aggregationPlan：To avoid redundancy, these three execution plans are set as subclasses of RawDataQueryPlan and set as variables in AlignByDevicePlan. If the query plan belongs to one of these three plans, the field is assigned and saved.
Before explaining the specific implementation process, a relatively complete example is given first, and the following explanation will be used in conjunction with this example.
@@ -82,7 +82,7 @@ First explain the meaning of some important fields in the `transformQuery ()` me
Next, introduce the calculation process of AlignByDevicePlan:
-1. Check whether the query type is one of three types of queries: groupByPlan, fillQueryPlan, aggregationPlan. If it is, assign the corresponding variable and change the query type of `AlignByDevicePlan`.
+1. Check whether the query type is one of three types of queries: groupByTimePlan, fillQueryPlan, aggregationPlan. If it is, assign the corresponding variable and change the query type of `AlignByDevicePlan`.
2. Iterate through the SELECT suffix path, and set an intermediate variable for each suffix path as `measurementSetOfGivenSuffix` to record all measurements corresponding to the suffix path. If the suffix path starts with single or double quotes, increase the value directly in `measurements` and note that its type is` Constant`.
3. Otherwise, the device list is stitched with the suffix path to obtain a complete path. If the spliced path does not exist, you need to further determine whether the measurement exists in other devices. If none, temporarily identify it as NonExist. If the subsequent device appears, measurement, the NonExist value is overridden.
4. If the path exists after splicing, it is proved that the measurement is of type `Exist`, and the consistency of the data type needs to be checked. If it is not satisfied, an error message is returned. If it is met, the measurement is recorded.
diff --git a/docs/SystemDesign/DataQuery/GroupByQuery.md b/docs/SystemDesign/DataQuery/GroupByQuery.md
index a7987fe..b11eaae 100644
--- a/docs/SystemDesign/DataQuery/GroupByQuery.md
+++ b/docs/SystemDesign/DataQuery/GroupByQuery.md
@@ -217,7 +217,7 @@ The downsampling query logic with value filtering conditions is mainly in the `G
This class has the following key fields:
* private List\<IReaderByTimestamp\> allDataReaderList
-* private GroupByPlan groupByPlan
+* private GroupByPlan groupByTimePlan
* private TimeGenerator timestampGenerator
* private long timestamp is used to cache timestamp for the next group by partition
* private boolean hasCachedTimestamp used to determine whether there is a timestamp cache for the next group by partition
@@ -258,3 +258,26 @@ for (int cnt = 1; cnt < timeStampFetchSize && timestampGenerator.hasNext(); cnt+
}
}
```
+
+
+## Aggregated query with level
+
+After down-frequency query, we could also to count the total number of points of
+
+each node at the given level in current Metadata Tree.
+
+The logic is in the `GroupByLevelDataSet` class.
+
+1. In the beginning, get the final paths group by level and the origin path index to final path.
+ > For example, we could get final path `root.sg1` by `root.sg1.d1.s0,root.sg1.d2.s1` and `level=1`.
+
+2. Then, get the down-frequency query result: RowRecord.
+
+3. Finally, merge each RowRecord to NewRecord, which has fields like <final path, count>.
+
+ > For example, we will get new RowRecord `<root.sg1,7>` by `<root.sg1.d1.s0, 3>, <root.sg1.d2.s1, 4>` and level=1.
+
+
+> Attention:
+> 1. only support count aggregation
+> 2. root's level == 0
\ No newline at end of file
diff --git a/docs/UserGuide/Operation Manual/DML Data Manipulation Language.md b/docs/UserGuide/Operation Manual/DML Data Manipulation Language.md
index f503770..113a6ca 100644
--- a/docs/UserGuide/Operation Manual/DML Data Manipulation Language.md
+++ b/docs/UserGuide/Operation Manual/DML Data Manipulation Language.md
@@ -263,6 +263,69 @@ We will get the result like following:
| 40 | 5 |
+### Down-Frequency Aggregate Query with Level Clause
+
+Level could be defined to show count the number of points of each node at the given level in current Metadata Tree.
+
+This could be used to query the number of points under each device.
+
+The SQL statement is:
+
+This means aggregate query only with path by level.
+
+```
+select count(status) from root.ln.wf01.wt01 group by level=1;
+```
+
+
+| Time | count(root.ln) |
+| ------ | -------------- |
+| 0 | 7 |
+
+
+```
+select count(status) from root.ln.wf01.wt01 group by level=2;
+```
+
+| Time | count(root.ln.wf01) | count(root.ln.wf02) |
+| ------ | ------------------- | ------------------- |
+| 0 | 4 | 3 |
+
+
+We can also get down-frequency agggate query by level.
+
+```
+select count(status) from root.ln.wf01.wt01 group by ([0,20),3ms), level=1;
+```
+
+
+| Time | count(root.ln) |
+| ------ | -------------- |
+| 0 | 1 |
+| 3 | 0 |
+| 6 | 0 |
+| 9 | 1 |
+| 12 | 3 |
+| 15 | 0 |
+| 18 | 0 |
+
+Down-frequency aggregate query with sliding step and by level.
+
+```
+select count(status) from root.ln.wf01.wt01 group by ([0,20),2ms,3ms), level=1;
+```
+
+
+| Time | count(root.ln) |
+| ------ | -------------- |
+| 0 | 1 |
+| 3 | 0 |
+| 6 | 0 |
+| 9 | 0 |
+| 12 | 2 |
+| 15 | 0 |
+| 18 | 0 |
+
#### Down-Frequency Aggregate Query with Fill Clause
In group by fill, sliding step is not supported in group by clause
diff --git a/docs/zh/SystemDesign/DataQuery/AlignByDeviceQuery.md b/docs/zh/SystemDesign/DataQuery/AlignByDeviceQuery.md
index 8043ba1..9be3469 100644
--- a/docs/zh/SystemDesign/DataQuery/AlignByDeviceQuery.md
+++ b/docs/zh/SystemDesign/DataQuery/AlignByDeviceQuery.md
@@ -40,7 +40,7 @@ AlignByDevicePlan 即按设备对齐查询对应的表结构为：
- `Map<String, TSDataType> measurementDataTypeMap`：AlignByDevicePlan 要求不同设备的同名 sensor 数据类型一致，该字段是一个 `measurementName -> dataType` 的 Map 结构，用来验证同名 sensor 的数据类型一致性。如 `root.sg.d1.s1` 和 `root.sg.d2.s1` 应该是同一数据类型。
- `enum MeasurementType`：记录三种 measurement 类型。在任何设备中都不存在的 measurement 为 `NonExist` 类型；有单引号或双引号的 measurement 为 `Constant` 类型；存在的 measurement 为 `Exist` 类型。
- `Map<String, MeasurementType> measurementTypeMap`: 该字段是一个 `measureName -> measurementType` 的 Map 结构，用来记录查询中所有 measurement 的类型。
-- groupByPlan, fillQueryPlan, aggregationPlan：为了避免冗余，这三个执行计划被设定为 RawDataQueryPlan 的子类，而在 AlignByDevicePlan 中被设置为变量。如果查询计划属于这三个计划中的一种，则该字段会被赋值并保存。
+- groupByTimePlan, fillQueryPlan, aggregationPlan：为了避免冗余，这三个执行计划被设定为 RawDataQueryPlan 的子类，而在 AlignByDevicePlan 中被设置为变量。如果查询计划属于这三个计划中的一种，则该字段会被赋值并保存。
在进行具体实现过程的讲解前，先给出一个覆盖较为完整的例子，下面的解释过程中将结合该示例进行说明。
@@ -82,7 +82,7 @@ SELECT s1, "1", *, s2, s5 FROM root.sg.d1, root.sg.* WHERE time = 1 AND s1 < 25
接下来介绍 AlignByDevicePlan 的计算过程：
-1. 检查查询类型是否为 groupByPlan, fillQueryPlan, aggregationPlan 这三类查询中的一种，如果是则对相应的变量进行赋值，并更改 `AlignByDevicePlan` 的查询类型。
+1. 检查查询类型是否为 groupByTimePlan, fillQueryPlan, aggregationPlan 这三类查询中的一种，如果是则对相应的变量进行赋值，并更改 `AlignByDevicePlan` 的查询类型。
2. 遍历 SELECT 后缀路径，对每一个后缀路径设置一个中间变量为 `measurementSetOfGivenSuffix`，用来记录该后缀路径对应的所有 measurement。如果后缀路径以单引号或双引号开头，则直接在 `measurements` 中增加该值，并记录其类型为 `Constant` 类型。
3. 否则将设备列表与该后缀路径拼接，得到完整的路径，如果拼接后的路径不存在，需要进一步判断该 measurement 是否在其它设备中存在，如果都没有则暂时识别为 `NonExist`，如果后续出现设备存在该 measurement，则覆盖 `NonExist` 值为 `Exist`。
4. 如果拼接后路径存在，则证明 measurement 是 `Exist` 类型，需要检验数据类型的一致性，不满足返回错误信息，满足则记录下该 Measurement，对 `measurementSetOfGivenSuffix` 等进行更新。
diff --git a/docs/zh/SystemDesign/DataQuery/GroupByQuery.md b/docs/zh/SystemDesign/DataQuery/GroupByQuery.md
index ea308c9..0c12398 100644
--- a/docs/zh/SystemDesign/DataQuery/GroupByQuery.md
+++ b/docs/zh/SystemDesign/DataQuery/GroupByQuery.md
@@ -219,7 +219,7 @@ if (batchData.getMaxTimestamp() >= curEndTime) {
private List<IReaderByTimestamp> allDataReaderList
```
-* private GroupByPlan groupByPlan
+* private GroupByPlan groupByTimePlan
* private TimeGenerator timestampGenerator
@@ -265,4 +265,24 @@ for (int cnt = 1; cnt < timeStampFetchSize && timestampGenerator.hasNext(); cnt+
break;
}
}
-```
\ No newline at end of file
+```
+
+## 使用Level来汇总降采样的总点数
+
+降采样后，我们也可以使用level关键字来进一步汇总点数。
+
+这个逻辑在 `GroupByLevelDataSet`类里。
+
+1. 首先，把所有涉及到的时序按level来进行汇集，最后的路径。
+ > 例如把root.sg1.d1.s0,root.sg1.d2.s1按level=1汇集成root.sg1。
+
+2. 然后调用上述的降采样逻辑求出所有时序的总点数信息，这个会返回RowRecord数据结构。
+
+3. 最后，把降采样返回的RowRecord按上述的final paths，进行累加，组合成新的RowRecord。
+
+ > 例如，把《root.sg1.d1.s0,3》，《root.sg1.d2.s1，4》聚合成《root.sg1，7》
+
+
+> 注意:
+> 1. 这里只支持count操作
+> 2. root的层级level=0
\ No newline at end of file
diff --git a/docs/zh/UserGuide/Operation Manual/DML Data Manipulation Language.md b/docs/zh/UserGuide/Operation Manual/DML Data Manipulation Language.md
index 621beda..87cf119 100644
--- a/docs/zh/UserGuide/Operation Manual/DML Data Manipulation Language.md
+++ b/docs/zh/UserGuide/Operation Manual/DML Data Manipulation Language.md
@@ -295,6 +295,70 @@ SQL执行后的结果集如下所示：
| 35 | 3 |
| 40 | 5 |
+### 降采样后按Level聚合查询
+
+除此之外，还可以通过定义LEVEL来统计指定层级下的数据点个数。
+
+这可以用来查询不同层级下的数据点总个数
+
+语法是：
+
+这个可以用来查询某个路径下的总数据点数
+
+```
+select count(status) from root.ln.wf01.wt01 group by level=1;
+```
+
+
+| Time | count(root.ln) |
+| ------ | -------------- |
+| 0 | 7 |
+
+
+```
+select count(status) from root.ln.wf01.wt01 group by level=2;
+```
+
+| Time | count(root.ln.wf01) | count(root.ln.wf02) |
+| ------ | ------------------- | ------------------- |
+| 0 | 4 | 3 |
+
+
+也可以用来统计降采样后的数据点个数
+
+```
+select count(status) from root.ln.wf01.wt01 group by ([0,20),3ms), level=1;
+```
+
+
+| Time | count(root.ln) |
+| ------ | -------------- |
+| 0 | 1 |
+| 3 | 0 |
+| 6 | 0 |
+| 9 | 1 |
+| 12 | 3 |
+| 15 | 0 |
+| 18 | 0 |
+
+加上滑动Step的降采样后的结果也可以汇总
+
+```
+select count(status) from root.ln.wf01.wt01 group by ([0,20),2ms,3ms), level=1;
+```
+
+
+| Time | count(root.ln) |
+| ------ | -------------- |
+| 0 | 1 |
+| 3 | 0 |
+| 6 | 0 |
+| 9 | 0 |
+| 12 | 2 |
+| 15 | 0 |
+| 18 | 0 |
+
+
#### 降频聚合查询补空值
降频聚合出的各个时间段的结果，支持使用前值补空。
diff --git a/flink-tsfile-connector/src/main/java/org/apache/iotdb/flink/tsfile/RowRowRecordParser.java b/flink-tsfile-connector/src/main/java/org/apache/iotdb/flink/tsfile/RowRowRecordParser.java
index 0ea9bce..1a009fe 100644
--- a/flink-tsfile-connector/src/main/java/org/apache/iotdb/flink/tsfile/RowRowRecordParser.java
+++ b/flink-tsfile-connector/src/main/java/org/apache/iotdb/flink/tsfile/RowRowRecordParser.java
@@ -91,7 +91,7 @@ public class RowRowRecordParser implements RowRecordParser<Row>, ResultTypeQuery
/**
* Creates RowRowRecordParser from output RowTypeInfo and selected series in the RowRecord. The row field "time"
- * will be used to store the timestamp value. The other row fields store the values of the same field names of
+ * will be used to store the timestamp value. The other row fields store the values of the same field names of
* the RowRecord.
*
* @param outputRowTypeInfo The RowTypeInfo of the output row.
diff --git a/server/pom.xml b/server/pom.xml
index 351f6a6..85968df 100644
--- a/server/pom.xml
+++ b/server/pom.xml
@@ -118,6 +118,16 @@
<groupId>io.moquette</groupId>
<artifactId>moquette-broker</artifactId>
<version>0.13</version>
+ <exclusions>
+ <exclusion>
+ <groupId>log4j</groupId>
+ <artifactId>log4j</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-log4j12</artifactId>
+ </exclusion>
+ </exclusions>
</dependency>
<!-- for mocked test-->
<dependency>
diff --git a/server/src/main/antlr4/org/apache/iotdb/db/qp/strategy/SqlBase.g4 b/server/src/main/antlr4/org/apache/iotdb/db/qp/strategy/SqlBase.g4
index 08e9721..b9c30ea 100644
--- a/server/src/main/antlr4/org/apache/iotdb/db/qp/strategy/SqlBase.g4
+++ b/server/src/main/antlr4/org/apache/iotdb/db/qp/strategy/SqlBase.g4
@@ -192,7 +192,7 @@ fromClause
specialClause
: specialLimit
- | groupByClause specialLimit?
+ | groupByTimeClause specialLimit?
| groupByFillClause
| fillClause slimitClause? alignByDeviceClauseOrDisableAlign?
| alignByDeviceClauseOrDisableAlign
@@ -240,12 +240,19 @@ fillClause
: FILL LR_BRACKET typeClause (COMMA typeClause)* RR_BRACKET
;
-groupByClause
+groupByTimeClause
: GROUP BY LR_BRACKET
timeInterval
COMMA DURATION
(COMMA DURATION)?
RR_BRACKET
+ | GROUP BY LEVEL OPERATOR_EQ INT
+ | GROUP BY LR_BRACKET
+ timeInterval
+ COMMA DURATION
+ (COMMA DURATION)?
+ RR_BRACKET
+ COMMA LEVEL OPERATOR_EQ INT
;
groupByFillClause
diff --git a/server/src/main/java/org/apache/iotdb/db/auth/AuthorityChecker.java b/server/src/main/java/org/apache/iotdb/db/auth/AuthorityChecker.java
index ef3f1a4..ca9676a 100644
--- a/server/src/main/java/org/apache/iotdb/db/auth/AuthorityChecker.java
+++ b/server/src/main/java/org/apache/iotdb/db/auth/AuthorityChecker.java
@@ -121,7 +121,7 @@ public class AuthorityChecker {
case QUERY:
case SELECT:
case FILTER:
- case GROUPBY:
+ case GROUPBYTIME:
case SEQTABLESCAN:
case TABLESCAN:
case INDEXQUERY:
diff --git a/server/src/main/java/org/apache/iotdb/db/conf/adapter/IoTDBConfigDynamicAdapter.java b/server/src/main/java/org/apache/iotdb/db/conf/adapter/IoTDBConfigDynamicAdapter.java
index 5580e43..2e91b04 100644
--- a/server/src/main/java/org/apache/iotdb/db/conf/adapter/IoTDBConfigDynamicAdapter.java
+++ b/server/src/main/java/org/apache/iotdb/db/conf/adapter/IoTDBConfigDynamicAdapter.java
@@ -101,7 +101,7 @@ public class IoTDBConfigDynamicAdapter implements IDynamicAdapter {
* Static memory, includes all timeseries metadata, which equals to
* TIMESERIES_METADATA_SIZE_IN_BYTE * totalTimeseriesNum, the unit is byte.
* <p>
- * Currently， we think that static memory only consists of time series metadata information. We
+ * Currently, we think that static memory only consists of time series metadata information. We
* ignore the memory occupied by the tsfile information maintained in memory, because we think
* that this part occupies very little memory.
*/
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/executor/PlanExecutor.java b/server/src/main/java/org/apache/iotdb/db/qp/executor/PlanExecutor.java
index 831d269..dd84071 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/executor/PlanExecutor.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/executor/PlanExecutor.java
@@ -83,18 +83,8 @@ import org.apache.iotdb.db.qp.logical.Operator.OperatorType;
import org.apache.iotdb.db.qp.logical.sys.AuthorOperator;
import org.apache.iotdb.db.qp.logical.sys.AuthorOperator.AuthorType;
import org.apache.iotdb.db.qp.physical.PhysicalPlan;
-import org.apache.iotdb.db.qp.physical.crud.AggregationPlan;
-import org.apache.iotdb.db.qp.physical.crud.AlignByDevicePlan;
-import org.apache.iotdb.db.qp.physical.crud.DeletePlan;
-import org.apache.iotdb.db.qp.physical.crud.FillQueryPlan;
-import org.apache.iotdb.db.qp.physical.crud.GroupByFillPlan;
-import org.apache.iotdb.db.qp.physical.crud.GroupByPlan;
-import org.apache.iotdb.db.qp.physical.crud.InsertPlan;
-import org.apache.iotdb.db.qp.physical.crud.InsertTabletPlan;
-import org.apache.iotdb.db.qp.physical.crud.LastQueryPlan;
-import org.apache.iotdb.db.qp.physical.crud.QueryPlan;
-import org.apache.iotdb.db.qp.physical.crud.RawDataQueryPlan;
-import org.apache.iotdb.db.qp.physical.crud.UpdatePlan;
+import org.apache.iotdb.db.qp.physical.crud.*;
+import org.apache.iotdb.db.qp.physical.crud.GroupByTimePlan;
import org.apache.iotdb.db.qp.physical.sys.AlterTimeSeriesPlan;
import org.apache.iotdb.db.qp.physical.sys.AuthorPlan;
import org.apache.iotdb.db.qp.physical.sys.ClearCachePlan;
@@ -302,12 +292,12 @@ public class PlanExecutor implements IPlanExecutor {
if (queryPlan.getPaths() == null || queryPlan.getPaths().isEmpty()) {
// no time series are selected, return EmptyDataSet
return new EmptyDataSet();
- } else if (queryPlan instanceof GroupByFillPlan) {
- GroupByFillPlan groupByFillPlan = (GroupByFillPlan) queryPlan;
+ } else if (queryPlan instanceof GroupByFillTimePlan) {
+ GroupByFillTimePlan groupByFillPlan = (GroupByFillTimePlan) queryPlan;
return queryRouter.groupByFill(groupByFillPlan, context);
- } else if (queryPlan instanceof GroupByPlan) {
- GroupByPlan groupByPlan = (GroupByPlan) queryPlan;
- return queryRouter.groupBy(groupByPlan, context);
+ } else if (queryPlan instanceof GroupByTimePlan) {
+ GroupByTimePlan groupByTimePlan = (GroupByTimePlan) queryPlan;
+ return queryRouter.groupBy(groupByTimePlan, context);
} else if (queryPlan instanceof AggregationPlan) {
AggregationPlan aggregationPlan = (AggregationPlan) queryPlan;
queryDataSet = queryRouter.aggregate(aggregationPlan, context);
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/logical/Operator.java b/server/src/main/java/org/apache/iotdb/db/qp/logical/Operator.java
index 48c530e..0a849df 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/logical/Operator.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/logical/Operator.java
@@ -66,7 +66,7 @@ public abstract class Operator {
* If you want to add new OperatorType, you must add it in the last.
*/
public enum OperatorType {
- SFW, JOIN, UNION, FILTER, GROUPBY, ORDERBY, LIMIT, SELECT, SEQTABLESCAN, HASHTABLESCAN,
+ SFW, JOIN, UNION, FILTER, GROUPBYTIME, ORDERBY, LIMIT, SELECT, SEQTABLESCAN, HASHTABLESCAN,
MERGEJOIN, FILEREAD, NULL, TABLESCAN, UPDATE, INSERT, BATCHINSERT, DELETE, BASIC_FUNC, IN, QUERY, MERGEQUERY,
AGGREGATION, AUTHOR, FROM, FUNC, LOADDATA, METADATA, INDEX, INDEXQUERY, FILL,
SET_STORAGE_GROUP, CREATE_TIMESERIES, DELETE_TIMESERIES, CREATE_USER, DELETE_USER, MODIFY_PASSWORD,
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/logical/crud/QueryOperator.java b/server/src/main/java/org/apache/iotdb/db/qp/logical/crud/QueryOperator.java
index a621c6f..1166592 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/logical/crud/QueryOperator.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/logical/crud/QueryOperator.java
@@ -42,6 +42,9 @@ public class QueryOperator extends SFWOperator {
private Map<TSDataType, IFill> fillTypes;
private boolean isFill = false;
+ private boolean isGroupByLevel = false;
+ private int level = -1;
+
private int rowLimit = 0;
private int rowOffset = 0;
private int seriesLimit = 0;
@@ -71,12 +74,12 @@ public class QueryOperator extends SFWOperator {
this.fillTypes = fillTypes;
}
- public boolean isGroupBy() {
- return isGroupByTime;
+ public boolean isGroupByLevel() {
+ return isGroupByLevel;
}
- public void setGroupBy(boolean isGroupBy) {
- this.isGroupByTime = isGroupBy;
+ public void setGroupByLevel(boolean isGroupBy) {
+ this.isGroupByLevel = isGroupBy;
}
public boolean isLeftCRightO() {
@@ -174,4 +177,20 @@ public class QueryOperator extends SFWOperator {
public void setAlignByTime(boolean isAlignByTime) {
this.isAlignByTime = isAlignByTime;
}
+
+ public int getLevel() {
+ return level;
+ }
+
+ public void setLevel(int level) {
+ this.level = level;
+ }
+
+ public boolean isGroupByTime() {
+ return isGroupByTime;
+ }
+
+ public void setGroupByTime(boolean groupByTime) {
+ isGroupByTime = groupByTime;
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/AlignByDevicePlan.java b/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/AlignByDevicePlan.java
index 297c7b3..3edd8b2 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/AlignByDevicePlan.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/AlignByDevicePlan.java
@@ -35,7 +35,7 @@ public class AlignByDevicePlan extends QueryPlan {
// to record different kinds of measurement
private Map<String, MeasurementType> measurementTypeMap;
- private GroupByPlan groupByPlan;
+ private GroupByTimePlan groupByTimePlan;
private FillQueryPlan fillQueryPlan;
private AggregationPlan aggregationPlan;
@@ -85,13 +85,13 @@ public class AlignByDevicePlan extends QueryPlan {
this.measurementTypeMap = measurementTypeMap;
}
- public GroupByPlan getGroupByPlan() {
- return groupByPlan;
+ public GroupByTimePlan getGroupByTimePlan() {
+ return groupByTimePlan;
}
- public void setGroupByPlan(GroupByPlan groupByPlan) {
- this.groupByPlan = groupByPlan;
- this.setOperatorType(OperatorType.GROUPBY);
+ public void setGroupByTimePlan(GroupByTimePlan groupByTimePlan) {
+ this.groupByTimePlan = groupByTimePlan;
+ this.setOperatorType(OperatorType.GROUPBYTIME);
}
public FillQueryPlan getFillQueryPlan() {
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/GroupByFillPlan.java b/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/GroupByFillTimePlan.java
similarity index 93%
rename from server/src/main/java/org/apache/iotdb/db/qp/physical/crud/GroupByFillPlan.java
rename to server/src/main/java/org/apache/iotdb/db/qp/physical/crud/GroupByFillTimePlan.java
index 1d620c4..55400f7 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/GroupByFillPlan.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/GroupByFillTimePlan.java
@@ -24,11 +24,11 @@ import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import java.util.Map;
-public class GroupByFillPlan extends GroupByPlan {
+public class GroupByFillTimePlan extends GroupByTimePlan {
private Map<TSDataType, IFill> fillTypes;
- public GroupByFillPlan() {
+ public GroupByFillTimePlan() {
super();
setOperatorType(Operator.OperatorType.GROUP_BY_FILL);
}
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/GroupByPlan.java b/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/GroupByTimePlan.java
similarity index 86%
rename from server/src/main/java/org/apache/iotdb/db/qp/physical/crud/GroupByPlan.java
rename to server/src/main/java/org/apache/iotdb/db/qp/physical/crud/GroupByTimePlan.java
index e47d248..cc76465 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/GroupByPlan.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/GroupByTimePlan.java
@@ -20,7 +20,7 @@ package org.apache.iotdb.db.qp.physical.crud;
import org.apache.iotdb.db.qp.logical.Operator;
-public class GroupByPlan extends AggregationPlan {
+public class GroupByTimePlan extends AggregationPlan {
// [startTime, endTime)
private long startTime;
@@ -33,9 +33,11 @@ public class GroupByPlan extends AggregationPlan {
// if it is left close and right open interval
private boolean leftCRightO = true;
- public GroupByPlan() {
+ private boolean byTime = false;
+
+ public GroupByTimePlan() {
super();
- setOperatorType(Operator.OperatorType.GROUPBY);
+ setOperatorType(Operator.OperatorType.GROUPBYTIME);
}
public long getStartTime() {
@@ -77,4 +79,12 @@ public class GroupByPlan extends AggregationPlan {
public void setLeftCRightO(boolean leftCRightO) {
this.leftCRightO = leftCRightO;
}
+
+ public boolean isByTime() {
+ return byTime;
+ }
+
+ public void setByTime(boolean isByTime) {
+ this.byTime = isByTime;
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/QueryPlan.java b/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/QueryPlan.java
index 58b4c13..7f3b9bd 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/QueryPlan.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/QueryPlan.java
@@ -36,6 +36,8 @@ public abstract class QueryPlan extends PhysicalPlan {
private int rowLimit = 0;
private int rowOffset = 0;
+ private int level = -1;
+
private Map<String, Integer> pathToIndex = new HashMap<>();
public QueryPlan() {
@@ -99,4 +101,12 @@ public abstract class QueryPlan extends PhysicalPlan {
public Map<String, Integer> getPathToIndex() {
return pathToIndex;
}
+
+ public int getLevel() {
+ return level;
+ }
+
+ public void setLevel(int level) {
+ this.level = level;
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/strategy/LogicalGenerator.java b/server/src/main/java/org/apache/iotdb/db/qp/strategy/LogicalGenerator.java
index ff0acc2..41dfd6a 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/strategy/LogicalGenerator.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/strategy/LogicalGenerator.java
@@ -169,6 +169,7 @@ import org.apache.iotdb.tsfile.utils.StringContainer;
* This class is a listener and you can get an operator which is a logical plan.
*/
public class LogicalGenerator extends SqlBaseBaseListener {
+ private static Logger logger = LoggerFactory.getLogger(LogicalGenerator.class);
private RootOperator initializedOperator = null;
private ZoneId zoneId;
@@ -787,8 +788,9 @@ public class LogicalGenerator extends SqlBaseBaseListener {
@Override
public void enterGroupByFillClause(SqlBaseParser.GroupByFillClauseContext ctx) {
super.enterGroupByFillClause(ctx);
- queryOp.setGroupBy(true);
+ queryOp.setGroupByLevel(true);
queryOp.setFill(true);
+ queryOp.setGroupByTime(true);
queryOp.setLeftCRightO(ctx.timeInterval().LS_BRACKET() != null);
@@ -860,23 +862,32 @@ public class LogicalGenerator extends SqlBaseBaseListener {
}
@Override
- public void enterGroupByClause(GroupByClauseContext ctx) {
- super.enterGroupByClause(ctx);
- queryOp.setGroupBy(true);
- queryOp.setLeftCRightO(ctx.timeInterval().LS_BRACKET() != null);
- // parse timeUnit
- queryOp.setUnit(parseDuration(ctx.DURATION(0).getText()));
- queryOp.setSlidingStep(queryOp.getUnit());
- // parse sliding step
- if (ctx.DURATION().size() == 2) {
- queryOp.setSlidingStep(parseDuration(ctx.DURATION(1).getText()));
- if (queryOp.getSlidingStep() < queryOp.getUnit()) {
- throw new SQLParserException(
- "The third parameter sliding step shouldn't be smaller than the second parameter time interval.");
+ public void enterGroupByTimeClause(GroupByTimeClauseContext ctx) {
+ super.enterGroupByTimeClause(ctx);
+ queryOp.setGroupByLevel(true);
+
+ if (ctx.timeInterval() != null) {
+ queryOp.setGroupByTime(true);
+ queryOp.setLeftCRightO(ctx.timeInterval().LS_BRACKET() != null);
+ // parse timeUnit
+ queryOp.setUnit(parseDuration(ctx.DURATION(0).getText()));
+ queryOp.setSlidingStep(queryOp.getUnit());
+ // parse sliding step
+ if (ctx.DURATION().size() == 2) {
+ queryOp.setSlidingStep(parseDuration(ctx.DURATION(1).getText()));
+ if (queryOp.getSlidingStep() < queryOp.getUnit()) {
+ throw new SQLParserException(
+ "The third parameter sliding step shouldn't be smaller than the second parameter time interval.");
+ }
}
+
+ parseTimeInterval(ctx.timeInterval());
}
- parseTimeInterval(ctx.timeInterval());
+ if (ctx.INT() != null) {
+ logger.debug("group by level:" + ctx.INT().getText() );
+ queryOp.setLevel(Integer.parseInt(ctx.INT().getText()));
+ }
}
@Override
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/strategy/PhysicalGenerator.java b/server/src/main/java/org/apache/iotdb/db/qp/strategy/PhysicalGenerator.java
index 8a1d8e3..33d7e33 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/strategy/PhysicalGenerator.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/strategy/PhysicalGenerator.java
@@ -59,17 +59,9 @@ import org.apache.iotdb.db.qp.logical.sys.ShowDevicesOperator;
import org.apache.iotdb.db.qp.logical.sys.ShowTTLOperator;
import org.apache.iotdb.db.qp.logical.sys.ShowTimeSeriesOperator;
import org.apache.iotdb.db.qp.physical.PhysicalPlan;
-import org.apache.iotdb.db.qp.physical.crud.AggregationPlan;
-import org.apache.iotdb.db.qp.physical.crud.AlignByDevicePlan;
+import org.apache.iotdb.db.qp.physical.crud.*;
import org.apache.iotdb.db.qp.physical.crud.AlignByDevicePlan.MeasurementType;
-import org.apache.iotdb.db.qp.physical.crud.DeletePlan;
-import org.apache.iotdb.db.qp.physical.crud.FillQueryPlan;
-import org.apache.iotdb.db.qp.physical.crud.GroupByFillPlan;
-import org.apache.iotdb.db.qp.physical.crud.GroupByPlan;
-import org.apache.iotdb.db.qp.physical.crud.InsertPlan;
-import org.apache.iotdb.db.qp.physical.crud.LastQueryPlan;
-import org.apache.iotdb.db.qp.physical.crud.QueryPlan;
-import org.apache.iotdb.db.qp.physical.crud.RawDataQueryPlan;
+import org.apache.iotdb.db.qp.physical.crud.GroupByTimePlan;
import org.apache.iotdb.db.qp.physical.sys.AlterTimeSeriesPlan;
import org.apache.iotdb.db.qp.physical.sys.AuthorPlan;
import org.apache.iotdb.db.qp.physical.sys.ClearCachePlan;
@@ -96,11 +88,14 @@ import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.read.common.Path;
import org.apache.iotdb.tsfile.read.expression.IExpression;
import org.apache.iotdb.tsfile.utils.Pair;
+import org.slf4j.LoggerFactory;
+import org.slf4j.Logger;
/**
* Used to convert logical operator to physical plan
*/
public class PhysicalGenerator {
+ private static Logger logger = LoggerFactory.getLogger(PhysicalGenerator.class);
public PhysicalPlan transformToPhysicalPlan(Operator operator) throws QueryProcessException {
List<Path> paths;
@@ -288,40 +283,51 @@ public class PhysicalGenerator {
private PhysicalPlan transformQuery(QueryOperator queryOperator) throws QueryProcessException {
QueryPlan queryPlan;
- if (queryOperator.isGroupBy() && queryOperator.isFill()) {
- queryPlan = new GroupByFillPlan();
- ((GroupByFillPlan) queryPlan).setInterval(queryOperator.getUnit());
- ((GroupByFillPlan) queryPlan).setSlidingStep(queryOperator.getSlidingStep());
- ((GroupByFillPlan) queryPlan).setLeftCRightO(queryOperator.isLeftCRightO());
+ if (queryOperator.isGroupByLevel() && queryOperator.isFill()) {
+ queryPlan = new GroupByFillTimePlan();
+ ((GroupByFillTimePlan) queryPlan).setInterval(queryOperator.getUnit());
+ ((GroupByFillTimePlan) queryPlan).setSlidingStep(queryOperator.getSlidingStep());
+ ((GroupByFillTimePlan) queryPlan).setLeftCRightO(queryOperator.isLeftCRightO());
if (!queryOperator.isLeftCRightO()) {
- ((GroupByPlan) queryPlan).setStartTime(queryOperator.getStartTime() + 1);
- ((GroupByPlan) queryPlan).setEndTime(queryOperator.getEndTime() + 1);
+ ((GroupByTimePlan) queryPlan).setStartTime(queryOperator.getStartTime() + 1);
+ ((GroupByTimePlan) queryPlan).setEndTime(queryOperator.getEndTime() + 1);
} else {
- ((GroupByPlan) queryPlan).setStartTime(queryOperator.getStartTime());
- ((GroupByPlan) queryPlan).setEndTime(queryOperator.getEndTime());
+ ((GroupByTimePlan) queryPlan).setStartTime(queryOperator.getStartTime());
+ ((GroupByTimePlan) queryPlan).setEndTime(queryOperator.getEndTime());
}
- ((GroupByFillPlan) queryPlan)
+ ((GroupByFillTimePlan) queryPlan)
.setAggregations(queryOperator.getSelectOperator().getAggregations());
for (String aggregation : queryPlan.getAggregations()) {
if (!SQLConstant.LAST_VALUE.equals(aggregation)) {
throw new QueryProcessException("Group By Fill only support last_value function");
}
}
- ((GroupByFillPlan) queryPlan).setFillType(queryOperator.getFillTypes());
- } else if (queryOperator.isGroupBy()) {
- queryPlan = new GroupByPlan();
- ((GroupByPlan) queryPlan).setInterval(queryOperator.getUnit());
- ((GroupByPlan) queryPlan).setSlidingStep(queryOperator.getSlidingStep());
- ((GroupByPlan) queryPlan).setLeftCRightO(queryOperator.isLeftCRightO());
+ ((GroupByFillTimePlan) queryPlan).setFillType(queryOperator.getFillTypes());
+ ((GroupByFillTimePlan) queryPlan).setByTime(queryOperator.isGroupByTime());
+ } else if (queryOperator.isGroupByLevel()) {
+ queryPlan = new GroupByTimePlan();
+ ((GroupByTimePlan) queryPlan).setInterval(queryOperator.getUnit());
+ ((GroupByTimePlan) queryPlan).setSlidingStep(queryOperator.getSlidingStep());
+ ((GroupByTimePlan) queryPlan).setLeftCRightO(queryOperator.isLeftCRightO());
if (!queryOperator.isLeftCRightO()) {
- ((GroupByPlan) queryPlan).setStartTime(queryOperator.getStartTime() + 1);
- ((GroupByPlan) queryPlan).setEndTime(queryOperator.getEndTime() + 1);
+ ((GroupByTimePlan) queryPlan).setStartTime(queryOperator.getStartTime() + 1);
+ ((GroupByTimePlan) queryPlan).setEndTime(queryOperator.getEndTime() + 1);
} else {
- ((GroupByPlan) queryPlan).setStartTime(queryOperator.getStartTime());
- ((GroupByPlan) queryPlan).setEndTime(queryOperator.getEndTime());
+ ((GroupByTimePlan) queryPlan).setStartTime(queryOperator.getStartTime());
+ ((GroupByTimePlan) queryPlan).setEndTime(queryOperator.getEndTime());
}
- ((GroupByPlan) queryPlan)
+ ((GroupByTimePlan) queryPlan)
.setAggregations(queryOperator.getSelectOperator().getAggregations());
+ ((GroupByTimePlan) queryPlan).setLevel(queryOperator.getLevel());
+ ((GroupByTimePlan) queryPlan).setByTime(queryOperator.isGroupByTime());
+
+ if (queryOperator.getLevel() >= 0) {
+ for (int i = 0; i < queryOperator.getSelectOperator().getAggregations().size(); i++) {
+ if (!SQLConstant.COUNT.equals(queryOperator.getSelectOperator().getAggregations().get(i))) {
+ throw new QueryProcessException("group by level only support count");
+ }
+ }
+ }
} else if (queryOperator.isFill()) {
queryPlan = new FillQueryPlan();
FilterOperator timeFilter = queryOperator.getFilterOperator();
@@ -350,8 +356,8 @@ public class PhysicalGenerator {
} else if (queryOperator.isAlignByDevice()) {
// below is the core realization of ALIGN_BY_DEVICE sql logic
AlignByDevicePlan alignByDevicePlan = new AlignByDevicePlan();
- if (queryPlan instanceof GroupByPlan) {
- alignByDevicePlan.setGroupByPlan((GroupByPlan) queryPlan);
+ if (queryPlan instanceof GroupByTimePlan) {
+ alignByDevicePlan.setGroupByTimePlan((GroupByTimePlan) queryPlan);
} else if (queryPlan instanceof FillQueryPlan) {
alignByDevicePlan.setFillQueryPlan((FillQueryPlan) queryPlan);
} else if (queryPlan instanceof AggregationPlan) {
diff --git a/server/src/main/java/org/apache/iotdb/db/query/dataset/AlignByDeviceDataSet.java b/server/src/main/java/org/apache/iotdb/db/query/dataset/AlignByDeviceDataSet.java
index 5b4e9b0..581f405 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/dataset/AlignByDeviceDataSet.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/dataset/AlignByDeviceDataSet.java
@@ -34,7 +34,7 @@ import org.apache.iotdb.db.qp.physical.crud.AggregationPlan;
import org.apache.iotdb.db.qp.physical.crud.AlignByDevicePlan;
import org.apache.iotdb.db.qp.physical.crud.AlignByDevicePlan.MeasurementType;
import org.apache.iotdb.db.qp.physical.crud.FillQueryPlan;
-import org.apache.iotdb.db.qp.physical.crud.GroupByPlan;
+import org.apache.iotdb.db.qp.physical.crud.GroupByTimePlan;
import org.apache.iotdb.db.qp.physical.crud.RawDataQueryPlan;
import org.apache.iotdb.db.query.context.QueryContext;
import org.apache.iotdb.db.query.executor.IQueryRouter;
@@ -64,7 +64,7 @@ public class AlignByDeviceDataSet extends QueryDataSet {
private Map<String, MeasurementType> measurementTypeMap;
private Map<String, TSDataType> measurementDataTpeMap;
- private GroupByPlan groupByPlan;
+ private GroupByTimePlan groupByTimePlan;
private FillQueryPlan fillQueryPlan;
private AggregationPlan aggregationPlan;
private RawDataQueryPlan rawDataQueryPlan;
@@ -88,9 +88,9 @@ public class AlignByDeviceDataSet extends QueryDataSet {
this.measurementTypeMap = alignByDevicePlan.getMeasurementTypeMap();
switch (alignByDevicePlan.getOperatorType()) {
- case GROUPBY:
- this.dataSetType = DataSetType.GROUPBY;
- this.groupByPlan = alignByDevicePlan.getGroupByPlan();
+ case GROUPBYTIME:
+ this.dataSetType = DataSetType.GROUPBYTIME;
+ this.groupByTimePlan = alignByDevicePlan.getGroupByTimePlan();
break;
case AGGREGATION:
this.dataSetType = DataSetType.AGGREGATE;
@@ -134,7 +134,7 @@ public class AlignByDeviceDataSet extends QueryDataSet {
List<String> executeAggregations = new ArrayList<>();
for (String column : measurementDataTpeMap.keySet()) {
String measurement = column;
- if (dataSetType == DataSetType.GROUPBY || dataSetType == DataSetType.AGGREGATE) {
+ if (dataSetType == DataSetType.GROUPBYTIME || dataSetType == DataSetType.AGGREGATE) {
measurement = column.substring(column.indexOf('(') + 1, column.indexOf(')'));
if (measurementOfGivenDevice.contains(measurement)) {
executeAggregations.add(column.substring(0, column.indexOf('(')));
@@ -154,11 +154,11 @@ public class AlignByDeviceDataSet extends QueryDataSet {
try {
switch (dataSetType) {
- case GROUPBY:
- groupByPlan.setDeduplicatedPaths(executePaths);
- groupByPlan.setDeduplicatedDataTypes(tsDataTypes);
- groupByPlan.setDeduplicatedAggregations(executeAggregations);
- currentDataSet = queryRouter.groupBy(groupByPlan, context);
+ case GROUPBYTIME:
+ groupByTimePlan.setDeduplicatedPaths(executePaths);
+ groupByTimePlan.setDeduplicatedDataTypes(tsDataTypes);
+ groupByTimePlan.setDeduplicatedAggregations(executeAggregations);
+ currentDataSet = queryRouter.groupBy(groupByTimePlan, context);
break;
case AGGREGATE:
aggregationPlan.setDeduplicatedPaths(executePaths);
@@ -232,7 +232,7 @@ public class AlignByDeviceDataSet extends QueryDataSet {
}
private enum DataSetType {
- GROUPBY, AGGREGATE, FILL, QUERY
+ GROUPBYTIME, AGGREGATE, FILL, QUERY
}
}
diff --git a/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/GroupByEngineDataSet.java b/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/GroupByEngineDataSet.java
index eaf9d8a..df1a6cf 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/GroupByEngineDataSet.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/GroupByEngineDataSet.java
@@ -18,7 +18,7 @@
*/
package org.apache.iotdb.db.query.dataset.groupby;
-import org.apache.iotdb.db.qp.physical.crud.GroupByPlan;
+import org.apache.iotdb.db.qp.physical.crud.GroupByTimePlan;
import org.apache.iotdb.db.query.context.QueryContext;
import org.apache.iotdb.db.utils.TestOnly;
import org.apache.iotdb.tsfile.read.common.RowRecord;
@@ -49,14 +49,14 @@ public abstract class GroupByEngineDataSet extends QueryDataSet {
/**
* groupBy query.
*/
- public GroupByEngineDataSet(QueryContext context, GroupByPlan groupByPlan) {
- super(groupByPlan.getDeduplicatedPaths(), groupByPlan.getDeduplicatedDataTypes());
+ public GroupByEngineDataSet(QueryContext context, GroupByTimePlan groupByTimePlan) {
+ super(groupByTimePlan.getDeduplicatedPaths(), groupByTimePlan.getDeduplicatedDataTypes());
this.queryId = context.getQueryId();
- this.interval = groupByPlan.getInterval();
- this.slidingStep = groupByPlan.getSlidingStep();
- this.startTime = groupByPlan.getStartTime();
- this.endTime = groupByPlan.getEndTime();
- this.leftCRightO = groupByPlan.isLeftCRightO();
+ this.interval = groupByTimePlan.getInterval();
+ this.slidingStep = groupByTimePlan.getSlidingStep();
+ this.startTime = groupByTimePlan.getStartTime();
+ this.endTime = groupByTimePlan.getEndTime();
+ this.leftCRightO = groupByTimePlan.isLeftCRightO();
// init group by time partition
this.hasCachedTimeInterval = false;
this.curStartTime = this.startTime - slidingStep;
diff --git a/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/GroupByFillDataSet.java b/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/GroupByFillDataSet.java
index 3ad751a..d826f23 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/GroupByFillDataSet.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/GroupByFillDataSet.java
@@ -21,7 +21,7 @@ package org.apache.iotdb.db.query.dataset.groupby;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.exception.StorageEngineException;
import org.apache.iotdb.db.exception.query.QueryProcessException;
-import org.apache.iotdb.db.qp.physical.crud.GroupByFillPlan;
+import org.apache.iotdb.db.qp.physical.crud.GroupByFillTimePlan;
import org.apache.iotdb.db.query.context.QueryContext;
import org.apache.iotdb.db.query.executor.LastQueryExecutor;
import org.apache.iotdb.db.query.executor.fill.IFill;
@@ -49,7 +49,7 @@ public class GroupByFillDataSet extends QueryDataSet {
public GroupByFillDataSet(List<Path> paths, List<TSDataType> dataTypes,
GroupByEngineDataSet groupByEngineDataSet,
- Map<TSDataType, IFill> fillTypes, QueryContext context, GroupByFillPlan groupByFillPlan)
+ Map<TSDataType, IFill> fillTypes, QueryContext context, GroupByFillTimePlan groupByFillPlan)
throws StorageEngineException, IOException, QueryProcessException {
super(paths, dataTypes);
this.groupByEngineDataSet = groupByEngineDataSet;
@@ -57,9 +57,9 @@ public class GroupByFillDataSet extends QueryDataSet {
initPreviousParis(context, groupByFillPlan);
initLastTimeArray(context, groupByFillPlan);
}
-
- private void initPreviousParis(QueryContext context, GroupByFillPlan groupByFillPlan)
- throws StorageEngineException, IOException, QueryProcessException {
+
+ private void initPreviousParis(QueryContext context, GroupByFillTimePlan groupByFillPlan)
+ throws StorageEngineException, IOException, QueryProcessException {
previousValue = new Object[paths.size()];
for (int i = 0; i < paths.size(); i++) {
Path path = paths.get(i);
@@ -85,7 +85,7 @@ public class GroupByFillDataSet extends QueryDataSet {
}
}
- private void initLastTimeArray(QueryContext context, GroupByFillPlan groupByFillPlan)
+ private void initLastTimeArray(QueryContext context, GroupByFillTimePlan groupByFillPlan)
throws IOException, StorageEngineException, QueryProcessException {
lastTimeArray = new long[paths.size()];
Arrays.fill(lastTimeArray, Long.MAX_VALUE);
diff --git a/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/GroupByLevelDataSet.java b/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/GroupByLevelDataSet.java
new file mode 100644
index 0000000..b24c610
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/GroupByLevelDataSet.java
@@ -0,0 +1,208 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.query.dataset.groupby;
+
+import org.apache.iotdb.db.exception.StorageEngineException;
+import org.apache.iotdb.db.exception.query.QueryProcessException;
+import org.apache.iotdb.db.qp.physical.crud.GroupByTimePlan;
+import org.apache.iotdb.db.query.aggregation.AggregateResult;
+import org.apache.iotdb.db.query.context.QueryContext;
+import org.apache.iotdb.db.query.factory.AggregateResultFactory;
+import org.apache.iotdb.db.query.filter.TsFileFilter;
+import org.apache.iotdb.db.utils.FilePathUtils;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.read.common.Field;
+import org.apache.iotdb.tsfile.read.common.Path;
+import org.apache.iotdb.tsfile.read.common.RowRecord;
+import org.apache.iotdb.tsfile.read.filter.basic.Filter;
+import org.apache.iotdb.tsfile.read.query.dataset.QueryDataSet;
+import org.apache.iotdb.tsfile.utils.Binary;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.*;
+
+import static org.apache.iotdb.db.conf.IoTDBConstant.COLUMN_COLUMN;
+import static org.apache.iotdb.db.conf.IoTDBConstant.COLUMN_COUNT;
+
+public class GroupByLevelDataSet extends QueryDataSet {
+
+ private static final Logger logger = LoggerFactory
+ .getLogger(GroupByLevelDataSet.class);
+
+ private List<RowRecord> records = new ArrayList<>();
+ private int index = 0;
+
+ protected long queryId;
+ private GroupByTimePlan groupByTimePlan;
+ private QueryContext context;
+
+ private Map<Path, GroupByExecutor> pathExecutors = new HashMap<>();
+ private Map<Path, List<Integer>> resultIndexes = new HashMap<>();
+
+ public GroupByLevelDataSet(QueryContext context, GroupByTimePlan plan, GroupByEngineDataSet dataSet)
+ throws QueryProcessException, StorageEngineException, IOException {
+ this.queryId = context.getQueryId();
+ this.paths = plan.getPaths();
+ this.dataTypes = plan.getDataTypes();
+ this.groupByTimePlan = plan;
+ this.context = context;
+
+ if (logger.isDebugEnabled()) {
+ logger.debug("paths " + this.paths + " level:" + plan.getLevel());
+ }
+
+ Map<Integer, String> pathIndex = new HashMap<>();
+ Map<String, Long> finalPaths = FilePathUtils.getPathByLevel(plan.getPaths(), plan.getLevel(), pathIndex);
+
+ if (!plan.isByTime()) {
+ // does not has time interval,
+ // so we could group by time interval [MIN_VALUE, MAX_VALUE] to get the total number
+ initGroupByLevel();
+ RowRecord record = mergeRecordByPath(getRecordWithoutTimeInterval(), finalPaths, pathIndex);
+ if (record != null) {
+ records.add(record);
+ }
+ } else {
+ // get all records from GroupByDataSet, then we merge every record
+ if (logger.isDebugEnabled()) {
+ logger.debug("only group by level, paths:" + groupByTimePlan.getPaths());
+ }
+ while (dataSet != null && dataSet.hasNextWithoutConstraint()) {
+ RowRecord curRecord = mergeRecordByPath(dataSet.nextWithoutConstraint(), finalPaths, pathIndex);
+ if (curRecord != null) {
+ records.add(curRecord);
+ }
+ }
+ }
+
+ this.dataTypes = new ArrayList<>();
+ this.paths = new ArrayList<>();
+ for (int i = 0; i < finalPaths.size(); i++) {
+ this.dataTypes.add(TSDataType.INT64);
+ }
+ }
+
+ @Override
+ protected boolean hasNextWithoutConstraint() throws IOException {
+ return index < records.size();
+ }
+
+ @Override
+ protected RowRecord nextWithoutConstraint() {
+ return records.get(index++);
+ }
+
+ private void initGroupByLevel()
+ throws QueryProcessException, StorageEngineException {
+ // get all aggregation results, then we package them to one record
+ for (int i = 0; i < paths.size(); i++) {
+ Path path = paths.get(i);
+ if (!pathExecutors.containsKey(path)) {
+ //init GroupByExecutor
+ pathExecutors.put(path,
+ getGroupByExecutor(path, groupByTimePlan.getAllMeasurementsInDevice(path.getDevice()), dataTypes.get(i), this.context, null, null));
+ resultIndexes.put(path, new ArrayList<>());
+ } else {
+ throw new QueryProcessException("duplicated path found, path:" + path);
+ }
+ resultIndexes.get(path).add(i);
+ AggregateResult aggrResult = AggregateResultFactory
+ .getAggrResultByName(groupByTimePlan.getDeduplicatedAggregations().get(i), dataTypes.get(i));
+ pathExecutors.get(path).addAggregateResult(aggrResult);
+ }
+ }
+
+ private GroupByExecutor getGroupByExecutor(Path path, Set<String> allSensors, TSDataType dataType,
+ QueryContext context, Filter timeFilter, TsFileFilter fileFilter)
+ throws StorageEngineException, QueryProcessException {
+ return new LocalGroupByExecutor(path, allSensors, dataType, context, timeFilter, fileFilter);
+ }
+
+ private RowRecord getRecordWithoutTimeInterval()
+ throws IOException {
+ RowRecord record = new RowRecord(0);
+ AggregateResult[] fields = new AggregateResult[paths.size()];
+
+ try {
+ for (Map.Entry<Path, GroupByExecutor> pathToExecutorEntry : pathExecutors.entrySet()) {
+ GroupByExecutor executor = pathToExecutorEntry.getValue();
+ List<AggregateResult> aggregations = executor.calcResult(Long.MIN_VALUE, Long.MAX_VALUE);
+ for (int i = 0; i < aggregations.size(); i++) {
+ int resultIndex = resultIndexes.get(pathToExecutorEntry.getKey()).get(i);
+ fields[resultIndex] = aggregations.get(i);
+ }
+ }
+ } catch (QueryProcessException e) {
+ logger.error("GroupByWithoutValueFilterDataSet execute has error", e);
+ throw new IOException(e.getMessage(), e);
+ }
+
+ for (AggregateResult res : fields) {
+ if (res == null) {
+ record.addField(null);
+ continue;
+ }
+ record.addField(res.getResult(), res.getResultDataType());
+ }
+ return record;
+ }
+
+ /**
+ * merge the raw record by level, for example
+ * raw record [timestamp, root.sg1.d1.s0, root.sg1.d1.s1, root.sg1.d2.s2], level=1
+ * and newRecord data is [100, 1, 2]
+ * return [100, 3]
+ * @param newRecord
+ * @param finalPaths
+ * @param pathIndex
+ * @return
+ */
+ private RowRecord mergeRecordByPath(RowRecord newRecord,
+ Map<String, Long> finalPaths,
+ Map<Integer, String> pathIndex) {
+ if (paths.size() != newRecord.getFields().size()) {
+ logger.error("bad record, result size not equal path size");
+ return null;
+ }
+
+ // reset final paths
+ for (Map.Entry<String, Long> entry : finalPaths.entrySet()) {
+ entry.setValue(0L);
+ }
+
+ RowRecord tmpRecord = new RowRecord(newRecord.getTimestamp());
+
+ for (int i = 0; i < newRecord.getFields().size(); i++) {
+ if (newRecord.getFields().get(i) != null) {
+ finalPaths.put(pathIndex.get(i),
+ finalPaths.get(pathIndex.get(i)) + newRecord.getFields().get(i).getLongV());
+ }
+ }
+
+ for (Map.Entry<String, Long> entry : finalPaths.entrySet()) {
+ tmpRecord.addField(Field.getField(entry.getValue(), TSDataType.INT64));
+ }
+
+ return tmpRecord;
+ }
+
+}
diff --git a/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/GroupByWithValueFilterDataSet.java b/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/GroupByWithValueFilterDataSet.java
index ce1a39e..6393f3b 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/GroupByWithValueFilterDataSet.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/GroupByWithValueFilterDataSet.java
@@ -22,7 +22,7 @@ package org.apache.iotdb.db.query.dataset.groupby;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.exception.StorageEngineException;
import org.apache.iotdb.db.exception.query.QueryProcessException;
-import org.apache.iotdb.db.qp.physical.crud.GroupByPlan;
+import org.apache.iotdb.db.qp.physical.crud.GroupByTimePlan;
import org.apache.iotdb.db.qp.physical.crud.RawDataQueryPlan;
import org.apache.iotdb.db.query.aggregation.AggregateResult;
import org.apache.iotdb.db.query.context.QueryContext;
@@ -45,7 +45,7 @@ import java.util.List;
public class GroupByWithValueFilterDataSet extends GroupByEngineDataSet {
private List<IReaderByTimestamp> allDataReaderList;
- private GroupByPlan groupByPlan;
+ private GroupByTimePlan groupByTimePlan;
private TimeGenerator timestampGenerator;
/**
* cached timestamp for next group by partition.
@@ -67,15 +67,15 @@ public class GroupByWithValueFilterDataSet extends GroupByEngineDataSet {
/**
* constructor.
*/
- public GroupByWithValueFilterDataSet(QueryContext context, GroupByPlan groupByPlan)
+ public GroupByWithValueFilterDataSet(QueryContext context, GroupByTimePlan groupByTimePlan)
throws StorageEngineException, QueryProcessException {
- super(context, groupByPlan);
+ super(context, groupByTimePlan);
this.timeStampFetchSize = IoTDBDescriptor.getInstance().getConfig().getBatchSize();
- initGroupBy(context, groupByPlan);
+ initGroupBy(context, groupByTimePlan);
}
- public GroupByWithValueFilterDataSet(long queryId, GroupByPlan groupByPlan) {
- super(new QueryContext(queryId), groupByPlan);
+ public GroupByWithValueFilterDataSet(long queryId, GroupByTimePlan groupByTimePlan) {
+ super(new QueryContext(queryId), groupByTimePlan);
this.allDataReaderList = new ArrayList<>();
this.timeStampFetchSize = IoTDBDescriptor.getInstance().getConfig().getBatchSize();
}
@@ -83,14 +83,14 @@ public class GroupByWithValueFilterDataSet extends GroupByEngineDataSet {
/**
* init reader and aggregate function.
*/
- protected void initGroupBy(QueryContext context, GroupByPlan groupByPlan)
+ protected void initGroupBy(QueryContext context, GroupByTimePlan groupByTimePlan)
throws StorageEngineException, QueryProcessException {
- this.timestampGenerator = getTimeGenerator(groupByPlan.getExpression(), context, groupByPlan);
+ this.timestampGenerator = getTimeGenerator(groupByTimePlan.getExpression(), context, groupByTimePlan);
this.allDataReaderList = new ArrayList<>();
- this.groupByPlan = groupByPlan;
+ this.groupByTimePlan = groupByTimePlan;
for (int i = 0; i < paths.size(); i++) {
Path path = paths.get(i);
- allDataReaderList.add(getReaderByTime(path, groupByPlan, dataTypes.get(i), context, null));
+ allDataReaderList.add(getReaderByTime(path, groupByTimePlan, dataTypes.get(i), context, null));
}
}
@@ -116,8 +116,8 @@ public class GroupByWithValueFilterDataSet extends GroupByEngineDataSet {
List<AggregateResult> aggregateResultList = new ArrayList<>();
for (int i = 0; i < paths.size(); i++) {
aggregateResultList.add(AggregateResultFactory.getAggrResultByName(
- groupByPlan.getDeduplicatedAggregations().get(i),
- groupByPlan.getDeduplicatedDataTypes().get(i)));
+ groupByTimePlan.getDeduplicatedAggregations().get(i),
+ groupByTimePlan.getDeduplicatedDataTypes().get(i)));
}
long[] timestampArray = new long[timeStampFetchSize];
diff --git a/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/GroupByWithoutValueFilterDataSet.java b/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/GroupByWithoutValueFilterDataSet.java
index 264ba42..25db91a 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/GroupByWithoutValueFilterDataSet.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/GroupByWithoutValueFilterDataSet.java
@@ -21,7 +21,7 @@ package org.apache.iotdb.db.query.dataset.groupby;
import org.apache.iotdb.db.exception.StorageEngineException;
import org.apache.iotdb.db.exception.query.QueryProcessException;
-import org.apache.iotdb.db.qp.physical.crud.GroupByPlan;
+import org.apache.iotdb.db.qp.physical.crud.GroupByTimePlan;
import org.apache.iotdb.db.query.aggregation.AggregateResult;
import org.apache.iotdb.db.query.context.QueryContext;
import org.apache.iotdb.db.query.factory.AggregateResultFactory;
@@ -65,16 +65,16 @@ public class GroupByWithoutValueFilterDataSet extends GroupByEngineDataSet {
/**
* constructor.
*/
- public GroupByWithoutValueFilterDataSet(QueryContext context, GroupByPlan groupByPlan)
+ public GroupByWithoutValueFilterDataSet(QueryContext context, GroupByTimePlan groupByTimePlan)
throws StorageEngineException, QueryProcessException {
- super(context, groupByPlan);
+ super(context, groupByTimePlan);
- initGroupBy(context, groupByPlan);
+ initGroupBy(context, groupByTimePlan);
}
- protected void initGroupBy(QueryContext context, GroupByPlan groupByPlan)
+ protected void initGroupBy(QueryContext context, GroupByTimePlan groupByTimePlan)
throws StorageEngineException, QueryProcessException {
- IExpression expression = groupByPlan.getExpression();
+ IExpression expression = groupByTimePlan.getExpression();
Filter timeFilter = null;
if (expression != null) {
@@ -87,12 +87,12 @@ public class GroupByWithoutValueFilterDataSet extends GroupByEngineDataSet {
if (!pathExecutors.containsKey(path)) {
//init GroupByExecutor
pathExecutors.put(path,
- getGroupByExecutor(path, groupByPlan.getAllMeasurementsInDevice(path.getDevice()), dataTypes.get(i), context, timeFilter, null));
+ getGroupByExecutor(path, groupByTimePlan.getAllMeasurementsInDevice(path.getDevice()), dataTypes.get(i), context, timeFilter, null));
resultIndexes.put(path, new ArrayList<>());
}
resultIndexes.get(path).add(i);
AggregateResult aggrResult = AggregateResultFactory
- .getAggrResultByName(groupByPlan.getDeduplicatedAggregations().get(i), dataTypes.get(i));
+ .getAggrResultByName(groupByTimePlan.getDeduplicatedAggregations().get(i), dataTypes.get(i));
pathExecutors.get(path).addAggregateResult(aggrResult);
}
}
diff --git a/server/src/main/java/org/apache/iotdb/db/query/executor/IQueryRouter.java b/server/src/main/java/org/apache/iotdb/db/query/executor/IQueryRouter.java
index 9c1f7ec..9082900 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/executor/IQueryRouter.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/executor/IQueryRouter.java
@@ -45,7 +45,7 @@ public interface IQueryRouter {
/**
* Execute groupBy query.
*/
- QueryDataSet groupBy(GroupByPlan groupByPlan, QueryContext context)
+ QueryDataSet groupBy(GroupByTimePlan groupByTimePlan, QueryContext context)
throws QueryFilterOptimizationException, StorageEngineException,
QueryProcessException, IOException;
@@ -58,7 +58,7 @@ public interface IQueryRouter {
/**
* Execute group by fill query
*/
- QueryDataSet groupByFill(GroupByFillPlan groupByFillPlan, QueryContext context)
+ QueryDataSet groupByFill(GroupByFillTimePlan groupByFillPlan, QueryContext context)
throws QueryFilterOptimizationException, StorageEngineException,
QueryProcessException, IOException;
diff --git a/server/src/main/java/org/apache/iotdb/db/query/executor/QueryRouter.java b/server/src/main/java/org/apache/iotdb/db/query/executor/QueryRouter.java
index eb39d00..4823cc6 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/executor/QueryRouter.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/executor/QueryRouter.java
@@ -23,10 +23,7 @@ import org.apache.iotdb.db.exception.StorageEngineException;
import org.apache.iotdb.db.exception.query.QueryProcessException;
import org.apache.iotdb.db.qp.physical.crud.*;
import org.apache.iotdb.db.query.context.QueryContext;
-import org.apache.iotdb.db.query.dataset.groupby.GroupByEngineDataSet;
-import org.apache.iotdb.db.query.dataset.groupby.GroupByFillDataSet;
-import org.apache.iotdb.db.query.dataset.groupby.GroupByWithValueFilterDataSet;
-import org.apache.iotdb.db.query.dataset.groupby.GroupByWithoutValueFilterDataSet;
+import org.apache.iotdb.db.query.dataset.groupby.*;
import org.apache.iotdb.db.query.executor.fill.IFill;
import org.apache.iotdb.tsfile.exception.filter.QueryFilterOptimizationException;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
@@ -38,6 +35,8 @@ import org.apache.iotdb.tsfile.read.expression.impl.GlobalTimeExpression;
import org.apache.iotdb.tsfile.read.expression.util.ExpressionOptimizer;
import org.apache.iotdb.tsfile.read.filter.GroupByFilter;
import org.apache.iotdb.tsfile.read.query.dataset.QueryDataSet;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.util.List;
@@ -49,6 +48,8 @@ import java.util.Map;
*/
public class QueryRouter implements IQueryRouter {
+ private static Logger logger = LoggerFactory.getLogger(QueryRouter.class);
+
@Override
public QueryDataSet rawDataQuery(RawDataQueryPlan queryPlan, QueryContext context)
throws StorageEngineException, QueryProcessException {
@@ -112,47 +113,64 @@ public class QueryRouter implements IQueryRouter {
}
@Override
- public QueryDataSet groupBy(GroupByPlan groupByPlan, QueryContext context)
- throws QueryFilterOptimizationException, StorageEngineException, QueryProcessException {
- long unit = groupByPlan.getInterval();
- long slidingStep = groupByPlan.getSlidingStep();
- long startTime = groupByPlan.getStartTime();
- long endTime = groupByPlan.getEndTime();
+ public QueryDataSet groupBy(GroupByTimePlan groupByTimePlan, QueryContext context)
+ throws QueryFilterOptimizationException, StorageEngineException, QueryProcessException, IOException {
+
+ logger.debug("paths:" + groupByTimePlan.getPaths() + " level:" + groupByTimePlan.getLevel() + " byTime:" + groupByTimePlan.isByTime());
+
+ GroupByEngineDataSet dataSet = null;
+ if (groupByTimePlan.isByTime()) {
+ long unit = groupByTimePlan.getInterval();
+ long slidingStep = groupByTimePlan.getSlidingStep();
+ long startTime = groupByTimePlan.getStartTime();
+ long endTime = groupByTimePlan.getEndTime();
- IExpression expression = groupByPlan.getExpression();
- List<Path> selectedSeries = groupByPlan.getDeduplicatedPaths();
+ IExpression expression = groupByTimePlan.getExpression();
+ List<Path> selectedSeries = groupByTimePlan.getDeduplicatedPaths();
- GlobalTimeExpression timeExpression = new GlobalTimeExpression(
+ GlobalTimeExpression timeExpression = new GlobalTimeExpression(
new GroupByFilter(unit, slidingStep, startTime, endTime));
- if (expression == null) {
- expression = timeExpression;
- } else {
- expression = BinaryExpression.and(expression, timeExpression);
- }
+ if (expression == null) {
+ expression = timeExpression;
+ } else {
+ expression = BinaryExpression.and(expression, timeExpression);
+ }
- // optimize expression to an executable one
- IExpression optimizedExpression = ExpressionOptimizer.getInstance()
+ // optimize expression to an executable one
+ IExpression optimizedExpression = ExpressionOptimizer.getInstance()
.optimize(expression, selectedSeries);
- groupByPlan.setExpression(optimizedExpression);
+ groupByTimePlan.setExpression(optimizedExpression);
- if (optimizedExpression.getType() == ExpressionType.GLOBAL_TIME) {
- return getGroupByWithoutValueFilterDataSet(context, groupByPlan);
- } else {
- return getGroupByWithValueFilterDataSet(context, groupByPlan);
+ if (optimizedExpression.getType() == ExpressionType.GLOBAL_TIME) {
+ dataSet = getGroupByWithoutValueFilterDataSet(context, groupByTimePlan);
+ } else {
+ dataSet = getGroupByWithValueFilterDataSet(context, groupByTimePlan);
+ }
}
+
+ if (groupByTimePlan.getLevel() >= 0) {
+ return groupByLevelWithoutTimeIntervalDataSet(context, groupByTimePlan, dataSet);
+ }
+ return dataSet;
}
- protected GroupByWithoutValueFilterDataSet getGroupByWithoutValueFilterDataSet(QueryContext context, GroupByPlan plan)
+ protected GroupByWithoutValueFilterDataSet getGroupByWithoutValueFilterDataSet(QueryContext context, GroupByTimePlan plan)
throws StorageEngineException, QueryProcessException {
return new GroupByWithoutValueFilterDataSet(context, plan);
}
- protected GroupByWithValueFilterDataSet getGroupByWithValueFilterDataSet(QueryContext context, GroupByPlan plan)
+ protected GroupByWithValueFilterDataSet getGroupByWithValueFilterDataSet(QueryContext context, GroupByTimePlan plan)
throws StorageEngineException, QueryProcessException {
return new GroupByWithValueFilterDataSet(context, plan);
}
+ protected GroupByLevelDataSet groupByLevelWithoutTimeIntervalDataSet(QueryContext context, GroupByTimePlan plan,
+ GroupByEngineDataSet dataSet)
+ throws StorageEngineException, QueryProcessException, IOException {
+ return new GroupByLevelDataSet(context, plan, dataSet);
+ }
+
@Override
public QueryDataSet fill(FillQueryPlan fillQueryPlan, QueryContext context)
throws StorageEngineException, QueryProcessException, IOException {
@@ -174,7 +192,7 @@ public class QueryRouter implements IQueryRouter {
}
@Override
- public QueryDataSet groupByFill(GroupByFillPlan groupByFillPlan, QueryContext context)
+ public QueryDataSet groupByFill(GroupByFillTimePlan groupByFillPlan, QueryContext context)
throws QueryFilterOptimizationException, StorageEngineException, QueryProcessException, IOException {
GroupByEngineDataSet groupByEngineDataSet = (GroupByEngineDataSet) groupBy(groupByFillPlan, context);
return new GroupByFillDataSet(groupByFillPlan.getDeduplicatedPaths(), groupByFillPlan.getDeduplicatedDataTypes(),
@@ -187,5 +205,4 @@ public class QueryRouter implements IQueryRouter {
LastQueryExecutor lastQueryExecutor = new LastQueryExecutor(lastQueryPlan);
return lastQueryExecutor.execute(context, lastQueryPlan);
}
-
}
diff --git a/server/src/main/java/org/apache/iotdb/db/service/TSServiceImpl.java b/server/src/main/java/org/apache/iotdb/db/service/TSServiceImpl.java
index 2e5412f..0acb9de 100644
--- a/server/src/main/java/org/apache/iotdb/db/service/TSServiceImpl.java
+++ b/server/src/main/java/org/apache/iotdb/db/service/TSServiceImpl.java
@@ -19,20 +19,14 @@
package org.apache.iotdb.db.service;
import static org.apache.iotdb.db.conf.IoTDBConfig.PATH_PATTERN;
+import static org.apache.iotdb.db.qp.logical.Operator.OperatorType.GROUPBYTIME;
import static org.apache.iotdb.db.qp.physical.sys.ShowPlan.ShowContentType.TIMESERIES;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.sql.SQLException;
import java.time.ZoneId;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Collections;
-import java.util.HashSet;
-import java.util.LinkedHashSet;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
+import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicLong;
import java.util.stream.Collectors;
@@ -79,6 +73,7 @@ import org.apache.iotdb.db.query.dataset.NonAlignEngineDataSet;
import org.apache.iotdb.db.query.dataset.RawQueryDataSetWithoutValueFilter;
import org.apache.iotdb.db.tools.watermark.GroupedLSBWatermarkEncoder;
import org.apache.iotdb.db.tools.watermark.WatermarkEncoder;
+import org.apache.iotdb.db.utils.FilePathUtils;
import org.apache.iotdb.db.utils.QueryDataSetUtils;
import org.apache.iotdb.db.utils.SchemaUtils;
import org.apache.iotdb.rpc.RpcUtils;
@@ -549,7 +544,7 @@ public class TSServiceImpl implements TSIService.Iface, ServerContext {
if (plan.getOperatorType() == OperatorType.FILL) {
throw new QueryProcessException("Fill doesn't support disable align clause.");
}
- if (plan.getOperatorType() == OperatorType.GROUPBY) {
+ if (plan.getOperatorType() == GROUPBYTIME) {
throw new QueryProcessException("Group by doesn't support disable align clause.");
}
}
@@ -698,6 +693,12 @@ public class TSServiceImpl implements TSIService.Iface, ServerContext {
// Last Query should return different respond instead of the static one
// because the query dataset and query id is different although the header of last query is same.
return StaticResps.LAST_RESP.deepCopy();
+ } else if (plan.getOperatorType() == GROUPBYTIME && plan.getLevel() >= 0) {
+ Map<String, Long> finalPaths = FilePathUtils.getPathByLevel(plan.getPaths(), plan.getLevel(), null);
+ for (Map.Entry<String, Long> entry : finalPaths.entrySet()) {
+ respColumns.add("count(" + entry.getKey() + ")");
+ columnsTypes.add(TSDataType.INT64.toString());
+ }
} else {
getWideQueryHeaders(plan, respColumns, columnsTypes);
resp.setColumnNameIndexMap(plan.getPathToIndex());
@@ -718,17 +719,17 @@ public class TSServiceImpl implements TSIService.Iface, ServerContext {
switch (plan.getOperatorType()) {
case QUERY:
case FILL:
- for (Path p : paths) {
- if (p.getAlias() != null) {
- respColumns.add(p.getFullPathWithAlias());
+ for (Path path : paths) {
+ if (path.getAlias() != null) {
+ respColumns.add(path.getFullPathWithAlias());
} else {
- respColumns.add(p.getFullPath());
+ respColumns.add(path.getFullPath());
}
}
seriesTypes = getSeriesTypesByString(respColumns, null);
break;
case AGGREGATION:
- case GROUPBY:
+ case GROUPBYTIME:
case GROUP_BY_FILL:
List<String> aggregations = plan.getAggregations();
if (aggregations.size() != paths.size()) {
diff --git a/server/src/main/java/org/apache/iotdb/db/utils/FilePathUtils.java b/server/src/main/java/org/apache/iotdb/db/utils/FilePathUtils.java
index 5e9297a..3c54541 100644
--- a/server/src/main/java/org/apache/iotdb/db/utils/FilePathUtils.java
+++ b/server/src/main/java/org/apache/iotdb/db/utils/FilePathUtils.java
@@ -19,7 +19,12 @@
package org.apache.iotdb.db.utils;
import java.io.File;
+import java.util.List;
+import java.util.Map;
+import java.util.TreeMap;
+
import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
+import org.apache.iotdb.tsfile.read.common.Path;
public class FilePathUtils {
@@ -46,4 +51,43 @@ public class FilePathUtils {
return resource.getFile().getAbsolutePath().split(PATH_SPLIT_STRING);
}
+ /**
+ * get paths from group by level, like root.sg1.d2.s0, root.sg1.d1.s1
+ * level=1, return [root.sg1, 0] and pathIndex turns to be [[0, root.sg1], [1, root.sg1]]
+ * @param rawPaths
+ * @param level
+ * @param pathIndex
+ * @return
+ */
+ public static Map<String, Long> getPathByLevel(List<Path> rawPaths, int level, Map<Integer, String> pathIndex) {
+ // pathGroupByLevel -> count
+ Map<String, Long> finalPaths = new TreeMap<>();
+
+ int i = 0;
+ for (Path value : rawPaths) {
+ String[] tmpPath = value.getFullPath().split("\\.");
+
+ String key;
+ if (tmpPath.length <= level) {
+ key = value.getFullPath();
+ } else {
+ StringBuilder path = new StringBuilder();
+ for (int k = 0; k <= level; k++) {
+ if (k == 0) {
+ path.append(tmpPath[k]);
+ } else {
+ path.append(".").append(tmpPath[k]);
+ }
+ }
+ key = path.toString();
+ }
+ finalPaths.putIfAbsent(key, 0L);
+ if (pathIndex != null) {
+ pathIndex.put(i++, key);
+ }
+ }
+
+ return finalPaths;
+ }
+
}
diff --git a/server/src/test/java/org/apache/iotdb/db/qp/PlannerTest.java b/server/src/test/java/org/apache/iotdb/db/qp/PlannerTest.java
index 772c594..e4396a0 100644
--- a/server/src/test/java/org/apache/iotdb/db/qp/PlannerTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/qp/PlannerTest.java
@@ -127,7 +127,7 @@ public class PlannerTest {
String groupbyStatement = "select sum(*) from root.vehicle where root.vehicle.device1.sensor1 > 50 group by ([100,1100), 20ms)";
PhysicalPlan plan9 = processor.parseSQLToPhysicalPlan(groupbyStatement);
- assertEquals(OperatorType.GROUPBY, plan9.getOperatorType());
+ assertEquals(OperatorType.GROUPBYTIME, plan9.getOperatorType());
String fillStatement = "select sensor1 from root.vehicle.device1 where time = 50 Fill(int32[linear, 5m, 5m], boolean[previous, 5m])";
PhysicalPlan plan10 = processor.parseSQLToPhysicalPlan(fillStatement);
diff --git a/server/src/test/java/org/apache/iotdb/db/qp/plan/PhysicalPlanTest.java b/server/src/test/java/org/apache/iotdb/db/qp/plan/PhysicalPlanTest.java
index d5faacc..c424292 100644
--- a/server/src/test/java/org/apache/iotdb/db/qp/plan/PhysicalPlanTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/qp/plan/PhysicalPlanTest.java
@@ -136,7 +136,7 @@ public class PhysicalPlanTest {
if (!plan.isQuery()) {
fail();
}
- GroupByPlan mergePlan = (GroupByPlan) plan;
+ GroupByTimePlan mergePlan = (GroupByTimePlan) plan;
assertEquals(3L, mergePlan.getInterval());
assertEquals(3L, mergePlan.getSlidingStep());
assertEquals(8L, mergePlan.getStartTime());
@@ -152,7 +152,7 @@ public class PhysicalPlanTest {
if (!plan.isQuery()) {
fail();
}
- GroupByPlan mergePlan = (GroupByPlan) plan;
+ GroupByTimePlan mergePlan = (GroupByTimePlan) plan;
assertEquals(111, mergePlan.getInterval());
}
@@ -165,7 +165,7 @@ public class PhysicalPlanTest {
if (!plan.isQuery()) {
fail();
}
- GroupByPlan mergePlan = (GroupByPlan) plan;
+ GroupByTimePlan mergePlan = (GroupByTimePlan) plan;
assertEquals(3 * 60 * 60 * 1000, mergePlan.getInterval());
assertEquals(24 * 60 * 60 * 1000, mergePlan.getSlidingStep());
assertEquals(1496379612000L, mergePlan.getStartTime());
@@ -239,10 +239,10 @@ public class PhysicalPlanTest {
if (!plan.isQuery()) {
fail();
}
- if (!(plan instanceof GroupByFillPlan)) {
+ if (!(plan instanceof GroupByFillTimePlan)) {
fail();
}
- GroupByFillPlan groupByFillPlan = (GroupByFillPlan) plan;
+ GroupByFillTimePlan groupByFillPlan = (GroupByFillTimePlan) plan;
assertEquals(3L, groupByFillPlan.getInterval());
assertEquals(3L, groupByFillPlan.getSlidingStep());
assertEquals(8L, groupByFillPlan.getStartTime());
@@ -268,10 +268,10 @@ public class PhysicalPlanTest {
if (!plan.isQuery()) {
fail();
}
- if (!(plan instanceof GroupByFillPlan)) {
+ if (!(plan instanceof GroupByFillTimePlan)) {
fail();
}
- GroupByFillPlan groupByFillPlan = (GroupByFillPlan) plan;
+ GroupByFillTimePlan groupByFillPlan = (GroupByFillTimePlan) plan;
assertEquals(3L, groupByFillPlan.getInterval());
assertEquals(3L, groupByFillPlan.getSlidingStep());
assertEquals(8L, groupByFillPlan.getStartTime());
@@ -299,10 +299,10 @@ public class PhysicalPlanTest {
if (!plan.isQuery()) {
fail();
}
- if (!(plan instanceof GroupByFillPlan)) {
+ if (!(plan instanceof GroupByFillTimePlan)) {
fail();
}
- GroupByFillPlan groupByFillPlan = (GroupByFillPlan) plan;
+ GroupByFillTimePlan groupByFillPlan = (GroupByFillTimePlan) plan;
assertEquals(3L, groupByFillPlan.getInterval());
assertEquals(3L, groupByFillPlan.getSlidingStep());
assertEquals(8L, groupByFillPlan.getStartTime());
diff --git a/server/src/test/java/org/apache/iotdb/db/query/dataset/GroupByLevelDataSetTest.java b/server/src/test/java/org/apache/iotdb/db/query/dataset/GroupByLevelDataSetTest.java
new file mode 100644
index 0000000..e473a5a
--- /dev/null
+++ b/server/src/test/java/org/apache/iotdb/db/query/dataset/GroupByLevelDataSetTest.java
@@ -0,0 +1,189 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iotdb.db.query.dataset;
+
+import org.apache.iotdb.db.exception.query.QueryProcessException;
+import org.apache.iotdb.db.metadata.MManager;
+import org.apache.iotdb.db.qp.Planner;
+import org.apache.iotdb.db.qp.executor.IPlanExecutor;
+import org.apache.iotdb.db.qp.executor.PlanExecutor;
+import org.apache.iotdb.db.qp.physical.crud.QueryPlan;
+import org.apache.iotdb.db.utils.EnvironmentUtils;
+import org.apache.iotdb.tsfile.read.query.dataset.QueryDataSet;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import static org.junit.Assert.*;
+
+public class GroupByLevelDataSetTest {
+ private IPlanExecutor queryExecutor = new PlanExecutor();
+ private Planner processor = new Planner();
+ private String[] sqls = {
+ "SET STORAGE GROUP TO root.vehicle",
+ "SET STORAGE GROUP TO root.test",
+ "CREATE TIMESERIES root.vehicle.d0.s0 WITH DATATYPE=INT32, ENCODING=RLE",
+ "CREATE TIMESERIES root.vehicle.d0.s1 WITH DATATYPE=TEXT, ENCODING=PLAIN",
+ "CREATE TIMESERIES root.test.d0.s0 WITH DATATYPE=INT32, ENCODING=RLE",
+ "CREATE TIMESERIES root.test.d0.s1 WITH DATATYPE=TEXT, ENCODING=PLAIN",
+ "insert into root.vehicle.d0(timestamp,s0) values(10,100)",
+ "insert into root.vehicle.d0(timestamp,s0,s1) values(12,101,'102')",
+ "insert into root.vehicle.d0(timestamp,s1) values(19,'103')",
+ "insert into root.vehicle.d0(timestamp,s0) values(20,1000)",
+ "insert into root.vehicle.d0(timestamp,s0,s1) values(22,1001,'1002')",
+ "insert into root.vehicle.d0(timestamp,s1) values(29,'1003')",
+ "insert into root.test.d0(timestamp,s0) values(10,106)",
+ "insert into root.test.d0(timestamp,s0,s1) values(14,107,'108')",
+ "insert into root.test.d0(timestamp,s1) values(16,'109')",
+ "insert into root.test.d0(timestamp,s0) values(30,1006)",
+ "insert into root.test.d0(timestamp,s0,s1) values(34,1007,'1008')",
+ "insert into root.test.d0(timestamp,s1) values(36,'1090')",
+ "insert into root.vehicle.d0(timestamp,s0) values(6,120)",
+ "insert into root.vehicle.d0(timestamp,s0,s1) values(38,121,'122')",
+ "insert into root.vehicle.d0(timestamp,s1) values(9,'123')",
+ "insert into root.vehicle.d0(timestamp,s0) values(16,128)",
+ "insert into root.vehicle.d0(timestamp,s0,s1) values(18,189,'198')",
+ "insert into root.vehicle.d0(timestamp,s1) values(99,'1234')",
+ "insert into root.test.d0(timestamp,s0) values(15,126)",
+ "insert into root.test.d0(timestamp,s0,s1) values(8,127,'128')",
+ "insert into root.test.d0(timestamp,s1) values(20,'129')",
+ "insert into root.test.d0(timestamp,s0) values(150,426)",
+ "insert into root.test.d0(timestamp,s0,s1) values(80,427,'528')",
+ "insert into root.test.d0(timestamp,s1) values(2,'1209')",
+ "insert into root.vehicle.d0(timestamp,s0) values(209,130)",
+ "insert into root.vehicle.d0(timestamp,s0,s1) values(206,131,'132')",
+ "insert into root.vehicle.d0(timestamp,s1) values(70,'33')",
+ "insert into root.test.d0(timestamp,s0) values(19,136)",
+ "insert into root.test.d0(timestamp,s0,s1) values(7,137,'138')",
+ "insert into root.test.d0(timestamp,s1) values(30,'139')",
+ "insert into root.test.d0(timestamp,s0) values(1900,1316)",
+ "insert into root.test.d0(timestamp,s0,s1) values(700,1307,'1038')",
+ "insert into root.test.d0(timestamp,s1) values(3000,'1309')"};
+
+ static {
+ MManager.getInstance().init();
+ }
+
+ public GroupByLevelDataSetTest() throws QueryProcessException {
+ }
+
+ @Before
+ public void setUp() throws Exception {
+ EnvironmentUtils.envSetUp();
+ for (String sql : sqls) {
+ queryExecutor.processNonQuery(processor.parseSQLToPhysicalPlan(sql));
+ }
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ EnvironmentUtils.cleanEnv();
+ }
+
+ @Test
+ public void testGroupByLevel() throws Exception {
+ QueryPlan queryPlan = (QueryPlan) processor
+ .parseSQLToPhysicalPlan("select count(s1) from root.test.* group by level=1");
+ QueryDataSet dataSet = queryExecutor.processQuery(queryPlan, EnvironmentUtils.TEST_QUERY_CONTEXT);
+
+ assertTrue(dataSet.hasNext());
+ assertEquals("0\t12", dataSet.next().toString());
+
+ // level 0
+ queryPlan = (QueryPlan) processor
+ .parseSQLToPhysicalPlan("select count(s1) from root.*.* group by level=0");
+ dataSet = queryExecutor.processQuery(queryPlan, EnvironmentUtils.TEST_QUERY_CONTEXT);
+
+ assertTrue(dataSet.hasNext());
+ assertEquals("0\t22", dataSet.next().toString());
+
+ // level large than series path
+ queryPlan = (QueryPlan) processor
+ .parseSQLToPhysicalPlan("select count(s1) from root.test.* group by level=5");
+ dataSet = queryExecutor.processQuery(queryPlan, EnvironmentUtils.TEST_QUERY_CONTEXT);
+
+ assertTrue(dataSet.hasNext());
+ assertEquals("0\t12", dataSet.next().toString());
+
+ // with time interval
+ queryPlan = (QueryPlan) processor
+ .parseSQLToPhysicalPlan("select count(s1) from root.test.* group by ([0,20), 1ms), level=1");
+ dataSet = queryExecutor.processQuery(queryPlan, EnvironmentUtils.TEST_QUERY_CONTEXT);
+
+ assertTrue(dataSet.hasNext());
+ assertEquals("0\t0", dataSet.next().toString());
+ assertTrue(dataSet.hasNext());
+ assertEquals("1\t0", dataSet.next().toString());
+ assertTrue(dataSet.hasNext());
+ assertEquals("2\t1", dataSet.next().toString());
+
+ queryPlan = (QueryPlan) processor
+ .parseSQLToPhysicalPlan("select count(s1) from root.test.* group by ([0,20), 1ms), level=0");
+ dataSet = queryExecutor.processQuery(queryPlan, EnvironmentUtils.TEST_QUERY_CONTEXT);
+
+ assertTrue(dataSet.hasNext());
+ assertEquals("0\t0", dataSet.next().toString());
+ assertTrue(dataSet.hasNext());
+ assertEquals("1\t0", dataSet.next().toString());
+ assertTrue(dataSet.hasNext());
+ assertEquals("2\t1", dataSet.next().toString());
+
+ queryPlan = (QueryPlan) processor
+ .parseSQLToPhysicalPlan("select count(s1) from root.test.* group by ([0,20), 1ms), level=6");
+ dataSet = queryExecutor.processQuery(queryPlan, EnvironmentUtils.TEST_QUERY_CONTEXT);
+
+ assertTrue(dataSet.hasNext());
+ assertEquals("0\t0", dataSet.next().toString());
+ assertTrue(dataSet.hasNext());
+ assertEquals("1\t0", dataSet.next().toString());
+ assertTrue(dataSet.hasNext());
+ assertEquals("2\t1", dataSet.next().toString());
+
+ // multi paths
+ queryPlan = (QueryPlan) processor
+ .parseSQLToPhysicalPlan("select count(s1) from root.test.*,root.vehicle.* group by ([0,20), 1ms), level=1");
+ dataSet = queryExecutor.processQuery(queryPlan, EnvironmentUtils.TEST_QUERY_CONTEXT);
+
+ assertTrue(dataSet.hasNext());
+ assertEquals("0\t0\t0", dataSet.next().toString());
+ assertTrue(dataSet.hasNext());
+ assertEquals("1\t0\t0", dataSet.next().toString());
+ assertTrue(dataSet.hasNext());
+ assertEquals("2\t1\t0", dataSet.next().toString());
+
+ // with sliding step
+ queryPlan = (QueryPlan) processor
+ .parseSQLToPhysicalPlan("select count(s1) from root.test.* group by ([0,20), 3ms, 10ms), level=6");
+ dataSet = queryExecutor.processQuery(queryPlan, EnvironmentUtils.TEST_QUERY_CONTEXT);
+
+ assertTrue(dataSet.hasNext());
+ assertEquals("0\t1", dataSet.next().toString());
+
+ // not count
+ try {
+ queryPlan = (QueryPlan) processor
+ .parseSQLToPhysicalPlan("select sum(s0) from root.test.* group by ([0,200), 1ms), level=6");
+ dataSet = queryExecutor.processQuery(queryPlan, EnvironmentUtils.TEST_QUERY_CONTEXT);
+ fail();
+ } catch (Exception e) {
+ assertEquals("group by level only support count", e.getMessage());
+ }
+ }
+
+}
diff --git a/server/src/test/java/org/apache/iotdb/db/query/executor/GroupByEngineDataSetTest.java b/server/src/test/java/org/apache/iotdb/db/query/executor/GroupByEngineDataSetTest.java
index 1b08468..96af121 100644
--- a/server/src/test/java/org/apache/iotdb/db/query/executor/GroupByEngineDataSetTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/query/executor/GroupByEngineDataSetTest.java
@@ -22,7 +22,7 @@ import java.io.IOException;
import java.util.ArrayList;
import org.apache.iotdb.db.exception.StorageEngineException;
import org.apache.iotdb.db.exception.query.PathException;
-import org.apache.iotdb.db.qp.physical.crud.GroupByPlan;
+import org.apache.iotdb.db.qp.physical.crud.GroupByTimePlan;
import org.apache.iotdb.db.query.aggregation.impl.CountAggrResult;
import org.apache.iotdb.db.query.dataset.groupby.GroupByEngineDataSet;
import org.apache.iotdb.db.query.dataset.groupby.GroupByWithValueFilterDataSet;
@@ -43,13 +43,13 @@ public class GroupByEngineDataSetTest {
long[] startTimeArray = {8, 13, 18, 23, 28};
long[] endTimeArray = {11, 16, 21, 26, 31};
- GroupByPlan groupByPlan = new GroupByPlan();
- groupByPlan.setInterval(unit);
- groupByPlan.setSlidingStep(slidingStep);
- groupByPlan.setStartTime(startTime);
- groupByPlan.setEndTime(endTime);
+ GroupByTimePlan groupByTimePlan = new GroupByTimePlan();
+ groupByTimePlan.setInterval(unit);
+ groupByTimePlan.setSlidingStep(slidingStep);
+ groupByTimePlan.setStartTime(startTime);
+ groupByTimePlan.setEndTime(endTime);
- GroupByEngineDataSet groupByEngine = new GroupByWithValueFilterDataSet(queryId, groupByPlan);
+ GroupByEngineDataSet groupByEngine = new GroupByWithValueFilterDataSet(queryId, groupByTimePlan);
int cnt = 0;
while (groupByEngine.hasNext()) {
Pair pair = groupByEngine.nextTimePartition();
@@ -72,12 +72,12 @@ public class GroupByEngineDataSetTest {
long[] startTimeArray = {8, 13, 18, 23, 28};
long[] endTimeArray = {11, 16, 21, 26, 31};
- GroupByPlan groupByPlan = new GroupByPlan();
- groupByPlan.setInterval(unit);
- groupByPlan.setSlidingStep(slidingStep);
- groupByPlan.setStartTime(startTime);
- groupByPlan.setEndTime(endTime);
- GroupByEngineDataSet groupByEngine = new GroupByWithValueFilterDataSet(queryId, groupByPlan);
+ GroupByTimePlan groupByTimePlan = new GroupByTimePlan();
+ groupByTimePlan.setInterval(unit);
+ groupByTimePlan.setSlidingStep(slidingStep);
+ groupByTimePlan.setStartTime(startTime);
+ groupByTimePlan.setEndTime(endTime);
+ GroupByEngineDataSet groupByEngine = new GroupByWithValueFilterDataSet(queryId, groupByTimePlan);
int cnt = 0;
while (groupByEngine.hasNext()) {
Pair pair = groupByEngine.nextTimePartition();
@@ -100,12 +100,12 @@ public class GroupByEngineDataSetTest {
long[] startTimeArray = {8, 11, 14, 17, 20};
long[] endTimeArray = {11, 14, 17, 20, 23};
- GroupByPlan groupByPlan = new GroupByPlan();
- groupByPlan.setInterval(unit);
- groupByPlan.setSlidingStep(slidingStep);
- groupByPlan.setStartTime(startTime);
- groupByPlan.setEndTime(endTime);
- GroupByEngineDataSet groupByEngine = new GroupByWithValueFilterDataSet(queryId, groupByPlan);
+ GroupByTimePlan groupByTimePlan = new GroupByTimePlan();
+ groupByTimePlan.setInterval(unit);
+ groupByTimePlan.setSlidingStep(slidingStep);
+ groupByTimePlan.setStartTime(startTime);
+ groupByTimePlan.setEndTime(endTime);
+ GroupByEngineDataSet groupByEngine = new GroupByWithValueFilterDataSet(queryId, groupByTimePlan);
int cnt = 0;
while (groupByEngine.hasNext()) {
Pair pair = groupByEngine.nextTimePartition();
@@ -128,12 +128,12 @@ public class GroupByEngineDataSetTest {
long[] startTimeArray = {8, 11, 14, 17, 20};
long[] endTimeArray = {11, 14, 17, 20, 23};
- GroupByPlan groupByPlan = new GroupByPlan();
- groupByPlan.setInterval(unit);
- groupByPlan.setSlidingStep(slidingStep);
- groupByPlan.setStartTime(startTime);
- groupByPlan.setEndTime(endTime);
- GroupByEngineDataSet groupByEngine = new GroupByWithValueFilterDataSet(queryId, groupByPlan);
+ GroupByTimePlan groupByTimePlan = new GroupByTimePlan();
+ groupByTimePlan.setInterval(unit);
+ groupByTimePlan.setSlidingStep(slidingStep);
+ groupByTimePlan.setStartTime(startTime);
+ groupByTimePlan.setEndTime(endTime);
+ GroupByEngineDataSet groupByEngine = new GroupByWithValueFilterDataSet(queryId, groupByTimePlan);
int cnt = 0;
while (groupByEngine.hasNext()) {
Pair pair = groupByEngine.nextTimePartition();
@@ -156,14 +156,14 @@ public class GroupByEngineDataSetTest {
long[] startTimeArray = {8, 11, 14, 17, 20, 23};
long[] endTimeArray = {11, 14, 17, 20, 23, 25};
- GroupByPlan groupByPlan = new GroupByPlan();
- groupByPlan.setInterval(unit);
- groupByPlan.setSlidingStep(slidingStep);
- groupByPlan.setStartTime(startTime);
- groupByPlan.setEndTime(endTime);
+ GroupByTimePlan groupByTimePlan = new GroupByTimePlan();
+ groupByTimePlan.setInterval(unit);
+ groupByTimePlan.setSlidingStep(slidingStep);
+ groupByTimePlan.setStartTime(startTime);
+ groupByTimePlan.setEndTime(endTime);
ArrayList<Object> aggrList = new ArrayList<>();
aggrList.add(new CountAggrResult());
- GroupByEngineDataSet groupByEngine = new GroupByWithValueFilterDataSet(queryId, groupByPlan);
+ GroupByEngineDataSet groupByEngine = new GroupByWithValueFilterDataSet(queryId, groupByTimePlan);
int cnt = 0;
while (groupByEngine.hasNext()) {
Pair pair = groupByEngine.nextTimePartition();
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/RowRecord.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/RowRecord.java
index fba67d5..5e1dd52 100644
--- a/tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/RowRecord.java
+++ b/tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/RowRecord.java
@@ -67,4 +67,12 @@ public class RowRecord {
public List<Field> getFields() {
return fields;
}
+
+ public void setFields(List<Field> fields) {
+ this.fields = fields;
+ }
+
+ public void setField(int index, Field field) {
+ this.fields.set(index, field);
+ }
}