You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by er...@apache.org on 2022/09/14 05:40:15 UTC
[iotdb] 01/02: [IOTDB-3784] tag aggregation
This is an automated email from the ASF dual-hosted git repository.
ericpai pushed a commit to branch feature/group_by_tag
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 070c0a1dd1e7cef03a087efbcec21c85427e219e
Author: jon <wa...@alibaba-inc.com>
AuthorDate: Wed Jul 13 14:19:43 2022 +0800
[IOTDB-3784] tag aggregation
---
.../org/apache/iotdb/db/qp/sql/IdentifierParser.g4 | 2 +-
.../org/apache/iotdb/db/qp/sql/IoTDBSqlParser.g4 | 9 +-
docs/UserGuide/Query-Data/Aggregate-Query.md | 174 ++++++++
docs/zh/UserGuide/Query-Data/Aggregate-Query.md | 168 +++++++-
.../db/it/aggregation/IoTDBTagAggregationIT.java | 445 +++++++++++++++++++++
.../schemaregion/rocksdb/RSchemaRegion.java | 119 +++---
.../iotdb/db/metadata/LocalSchemaProcessor.java | 6 +-
.../db/metadata/cache/DataNodeSchemaCache.java | 3 +
.../iotdb/db/metadata/cache/SchemaCacheEntry.java | 11 +-
.../iotdb/db/metadata/mtree/IMTreeBelowSG.java | 5 +-
.../db/metadata/mtree/MTreeBelowSGCachedImpl.java | 17 +-
.../db/metadata/mtree/MTreeBelowSGMemoryImpl.java | 38 +-
.../iotdb/db/metadata/path/MeasurementPath.java | 35 ++
.../db/metadata/schemaregion/ISchemaRegion.java | 10 +-
.../schemaregion/SchemaRegionMemoryImpl.java | 64 ++-
.../schemaregion/SchemaRegionSchemaFileImpl.java | 57 ++-
.../mpp/common/schematree/ClusterSchemaTree.java | 9 +-
.../schematree/node/SchemaMeasurementNode.java | 14 +
.../visitor/SchemaTreeMeasurementVisitor.java | 1 +
.../operator/process/TagAggregationOperator.java | 237 +++++++++++
.../operator/schema/SchemaFetchScanOperator.java | 7 +-
.../apache/iotdb/db/mpp/plan/analyze/Analysis.java | 45 +++
.../iotdb/db/mpp/plan/analyze/AnalyzeVisitor.java | 89 ++++-
.../db/mpp/plan/analyze/ClusterSchemaFetcher.java | 13 +-
.../iotdb/db/mpp/plan/analyze/ISchemaFetcher.java | 2 +
.../mpp/plan/analyze/StandaloneSchemaFetcher.java | 12 +-
.../iotdb/db/mpp/plan/parser/ASTVisitor.java | 37 ++
.../db/mpp/plan/planner/LogicalPlanBuilder.java | 93 ++++-
.../db/mpp/plan/planner/LogicalPlanVisitor.java | 9 +-
.../db/mpp/plan/planner/OperatorTreeGenerator.java | 71 +++-
.../planner/distribution/ExchangeNodeAdder.java | 5 +
.../plan/planner/distribution/SourceRewriter.java | 19 +
.../plan/planner/plan/node/PlanGraphPrinter.java | 31 ++
.../mpp/plan/planner/plan/node/PlanNodeType.java | 6 +-
.../db/mpp/plan/planner/plan/node/PlanVisitor.java | 5 +
.../node/metedata/read/SchemaFetchScanNode.java | 19 +-
.../planner/plan/node/process/GroupByTagNode.java | 358 +++++++++++++++++
.../plan/parameter/AggregationDescriptor.java | 2 +-
.../statement/component/GroupByTagComponent.java | 38 ++
.../db/mpp/plan/statement/crud/QueryStatement.java | 22 +
.../statement/internal/SchemaFetchStatement.java | 9 +-
.../qp/strategy/optimizer/ConcatPathOptimizer.java | 3 +-
.../apache/iotdb/db/qp/utils/WildcardsRemover.java | 8 +-
.../db/metadata/cache/DataNodeSchemaCacheTest.java | 22 +-
.../iotdb/db/metadata/mtree/MTreeBelowSGTest.java | 16 +-
.../db/metadata/path/MeasurementPathTest.java | 104 +++++
.../mpp/execution/operator/OperatorMemoryTest.java | 7 +-
.../schema/SchemaFetchScanOperatorTest.java | 3 +-
.../mpp/plan/analyze/FakePartitionFetcherImpl.java | 0
.../db/mpp/plan/analyze/FakeSchemaFetcherImpl.java | 9 +
.../db/mpp/plan/parser/StatementGeneratorTest.java | 16 +
.../iotdb/db/mpp/plan/plan/LogicalPlannerTest.java | 93 ++++-
.../metadata/read/SchemaFetchScanNodeTest.java | 7 +-
.../plan/node/process/GroupByTagNodeSerdeTest.java | 121 ++++++
.../iotdb/tsfile/utils/ReadWriteIOUtils.java | 18 +-
55 files changed, 2598 insertions(+), 145 deletions(-)
diff --git a/antlr/src/main/antlr4/org/apache/iotdb/db/qp/sql/IdentifierParser.g4 b/antlr/src/main/antlr4/org/apache/iotdb/db/qp/sql/IdentifierParser.g4
index d0052decd8..f68b8651c7 100644
--- a/antlr/src/main/antlr4/org/apache/iotdb/db/qp/sql/IdentifierParser.g4
+++ b/antlr/src/main/antlr4/org/apache/iotdb/db/qp/sql/IdentifierParser.g4
@@ -179,4 +179,4 @@ keyWords
| WITH
| WITHOUT
| PRIVILEGE_VALUE
- ;
\ No newline at end of file
+ ;
diff --git a/antlr/src/main/antlr4/org/apache/iotdb/db/qp/sql/IoTDBSqlParser.g4 b/antlr/src/main/antlr4/org/apache/iotdb/db/qp/sql/IoTDBSqlParser.g4
index 28851aee49..9ad02eb8a5 100644
--- a/antlr/src/main/antlr4/org/apache/iotdb/db/qp/sql/IoTDBSqlParser.g4
+++ b/antlr/src/main/antlr4/org/apache/iotdb/db/qp/sql/IoTDBSqlParser.g4
@@ -394,6 +394,7 @@ specialClause
| groupByTimeClause havingClause? orderByClause? specialLimit? #groupByTimeStatement
| groupByFillClause havingClause? orderByClause? specialLimit? #groupByFillStatement
| groupByLevelClause havingClause? orderByClause? specialLimit? #groupByLevelStatement
+ | groupByTagClause orderByClause? #groupByTagStatement
| fillClause orderByClause? specialLimit? #fillStatement
;
@@ -440,6 +441,8 @@ groupByTimeClause
: GROUP BY LR_BRACKET timeRange COMMA DURATION_LITERAL (COMMA DURATION_LITERAL)? fillClause? RR_BRACKET
| GROUP BY LR_BRACKET timeRange COMMA DURATION_LITERAL (COMMA DURATION_LITERAL)? RR_BRACKET
COMMA LEVEL operator_eq INTEGER_LITERAL (COMMA INTEGER_LITERAL)* fillClause?
+ | GROUP BY LR_BRACKET timeRange COMMA DURATION_LITERAL (COMMA DURATION_LITERAL)? RR_BRACKET
+ COMMA TAGS identifier (COMMA identifier)*
;
groupByFillClause
@@ -451,6 +454,10 @@ groupByLevelClause
: GROUP BY LEVEL operator_eq INTEGER_LITERAL (COMMA INTEGER_LITERAL)* fillClause?
;
+groupByTagClause
+ : GROUP BY TAGS identifier (COMMA identifier)*
+ ;
+
fillClause
: FILL LR_BRACKET (linearClause | previousClause | specificValueClause | previousUntilLastClause | oldTypeClause (COMMA oldTypeClause)*) RR_BRACKET
;
@@ -957,4 +964,4 @@ slimitClause
soffsetClause
: SOFFSET INTEGER_LITERAL
- ;
\ No newline at end of file
+ ;
diff --git a/docs/UserGuide/Query-Data/Aggregate-Query.md b/docs/UserGuide/Query-Data/Aggregate-Query.md
index 3cc502b4c4..d81f18c189 100644
--- a/docs/UserGuide/Query-Data/Aggregate-Query.md
+++ b/docs/UserGuide/Query-Data/Aggregate-Query.md
@@ -454,6 +454,180 @@ Total line number = 7
It costs 0.004s
```
+## Aggregation By Tags
+
+IotDB allows you to do aggregation query with the tags defined in timeseries through `GROUP BY TAGS` clause as well.
+
+Firstly, we can put these example data into IoTDB, which will be used in the following feature introduction.
+
+These are the temperature data of the workshops, which belongs to the factory `factory1` and locates in different cities. The time range is `[1000, 10000)`.
+
+The device node of the timeseries path is the ID of the device. The information of city and workshop are modelled in the tags `city` and `workshop`.
+The devices `d1` and `d2` belong to the workshop `d1` in `Beijing`.
+`d3` and `d4` belong to the workshop `w2` in `Beijing`.
+`d5` and `d6` belong to the workshop `w1` in `Shanghai`.
+`d7` belongs to the workshop `w2` in `Shanghai`.
+`d8` and `d9` are under maintenance, and don't belong to any workshops, so they have no tags.
+
+
+```SQL
+set storage group to root.factory1;
+create timeseries root.factory1.d1.temperature with datatype=FLOAT tags(city=Beijing, workshop=w1);
+create timeseries root.factory1.d2.temperature with datatype=FLOAT tags(city=Beijing, workshop=w1);
+create timeseries root.factory1.d3.temperature with datatype=FLOAT tags(city=Beijing, workshop=w2);
+create timeseries root.factory1.d4.temperature with datatype=FLOAT tags(city=Beijing, workshop=w2);
+create timeseries root.factory1.d5.temperature with datatype=FLOAT tags(city=Shanghai, workshop=w1);
+create timeseries root.factory1.d6.temperature with datatype=FLOAT tags(city=Shanghai, workshop=w1);
+create timeseries root.factory1.d7.temperature with datatype=FLOAT tags(city=Shanghai, workshop=w2);
+create timeseries root.factory1.d8.temperature with datatype=FLOAT;
+create timeseries root.factory1.d9.temperature with datatype=FLOAT;
+
+insert into root.factory1.d1(time, temperature) values(1000, 104.0);
+insert into root.factory1.d1(time, temperature) values(3000, 104.2);
+insert into root.factory1.d1(time, temperature) values(5000, 103.3);
+insert into root.factory1.d1(time, temperature) values(7000, 104.1);
+
+insert into root.factory1.d2(time, temperature) values(1000, 104.4);
+insert into root.factory1.d2(time, temperature) values(3000, 103.7);
+insert into root.factory1.d2(time, temperature) values(5000, 103.3);
+insert into root.factory1.d2(time, temperature) values(7000, 102.9);
+
+insert into root.factory1.d3(time, temperature) values(1000, 103.9);
+insert into root.factory1.d3(time, temperature) values(3000, 103.8);
+insert into root.factory1.d3(time, temperature) values(5000, 102.7);
+insert into root.factory1.d3(time, temperature) values(7000, 106.9);
+
+insert into root.factory1.d4(time, temperature) values(1000, 103.9);
+insert into root.factory1.d4(time, temperature) values(5000, 102.7);
+insert into root.factory1.d4(time, temperature) values(7000, 106.9);
+
+insert into root.factory1.d5(time, temperature) values(1000, 112.9);
+insert into root.factory1.d5(time, temperature) values(7000, 113.0);
+
+insert into root.factory1.d6(time, temperature) values(1000, 113.9);
+insert into root.factory1.d6(time, temperature) values(3000, 113.3);
+insert into root.factory1.d6(time, temperature) values(5000, 112.7);
+insert into root.factory1.d6(time, temperature) values(7000, 112.3);
+
+insert into root.factory1.d7(time, temperature) values(1000, 101.2);
+insert into root.factory1.d7(time, temperature) values(3000, 99.3);
+insert into root.factory1.d7(time, temperature) values(5000, 100.1);
+insert into root.factory1.d7(time, temperature) values(7000, 99.8);
+
+insert into root.factory1.d8(time, temperature) values(1000, 50.0);
+insert into root.factory1.d8(time, temperature) values(3000, 52.1);
+insert into root.factory1.d8(time, temperature) values(5000, 50.1);
+insert into root.factory1.d8(time, temperature) values(7000, 50.5);
+
+insert into root.factory1.d9(time, temperature) values(1000, 50.3);
+insert into root.factory1.d9(time, temperature) values(3000, 52.1);
+```
+
+### Aggregation query by one single tag
+
+If the user wants to know the average temperature of each workshop, he can query like this
+
+```SQL
+SELECT AVG(temperature) FROM root.sg.** GROUP BY TAGS city;
+```
+
+The query will calculate the average of the temperatures of those timeseries which have the same tag value of the key `city`.
+The results are
+
+```
++--------+------------------+
+| city| avg(temperature)|
++--------+------------------+
+| Beijing|104.04666697184244|
+|Shanghai|107.85000076293946|
+| NULL| 50.84999910990397|
++--------+------------------+
+Total line number = 3
+It costs 0.231s
+```
+
+From the results we can see that the differences between aggregation by tags query and aggregation by time or level query are:
+1. Aggregation query by tags will no longer remove wildcard to raw timeseries, but do the aggregation through the data of multiple timeseries, which have the same tag value.
+2. Except for the aggregate result column, the result set contains the key-value column of the grouped tag. The column name is the tag key, and the values in the column are tag values which present in the searched timeseries.
+If some searched timeseries doesn't have the grouped tag, a `NULL` value in the key-value column of the grouped tag will be presented, which means the aggregation of all the timeseries lacking the tagged key.
+
+### Aggregation query by multiple tags
+
+Except for the aggregation query by one single tag, aggregation query by multiple tags in a particular order is allowed as well.
+
+For example, a user wants to know the average temperature of the devices in each workshop.
+As the workshop names may be same in different city, it's not correct to aggregated by the tag `workshop` directly.
+So the aggregation by the tag `city` should be done first, and then by the tag `workshop`.
+
+SQL
+
+```SQL
+select avg(temperature) FROM root.factory1.** GROUP BY TAGS city, workshop;
+```
+
+The results
+
+```
++--------+--------+------------------+
+| city|workshop| avg(temperature)|
++--------+--------+------------------+
+| NULL| NULL| 50.84999910990397|
+|Shanghai| w1|113.01666768391927|
+| Beijing| w2| 104.4000004359654|
+|Shanghai| w2|100.10000038146973|
+| Beijing| w1|103.73750019073486|
++--------+--------+------------------+
+Total line number = 5
+It costs 0.027s
+```
+
+We can see that in a multiple tags aggregation query, the result set will output the key-value columns of all the grouped tag keys, which have the same order with the one in `GROUP BY TAGS`.
+
+### Downsampling Aggregation by tags based on Time Window
+
+Downsampling aggregation by time window is one of the most popular features in a time series database. IoTDB supports to do aggregation query by tags based on time window.
+
+For example, a user wants to know the average temperature of the devices in each workshop, in every 5 seconds, in the range of time `[1000, 10000)`.
+
+SQL
+
+```SQL
+select avg(temperature) from root.factory1.** group by ([1000, 10000), 5s), tags city, workshop;
+```
+
+The results
+
+```
++-----------------------------+--------+--------+------------------+
+| Time| city|workshop| avg(temperature)|
++-----------------------------+--------+--------+------------------+
+|1970-01-01T08:00:01.000+08:00| NULL| NULL| 50.91999893188476|
+|1970-01-01T08:00:01.000+08:00|Shanghai| w1|113.20000076293945|
+|1970-01-01T08:00:01.000+08:00| Beijing| w2| 103.4|
+|1970-01-01T08:00:01.000+08:00|Shanghai| w2| 100.1999994913737|
+|1970-01-01T08:00:01.000+08:00| Beijing| w1|103.81666692097981|
+|1970-01-01T08:00:06.000+08:00| NULL| NULL| 50.5|
+|1970-01-01T08:00:06.000+08:00|Shanghai| w1| 112.6500015258789|
+|1970-01-01T08:00:06.000+08:00| Beijing| w2| 106.9000015258789|
+|1970-01-01T08:00:06.000+08:00|Shanghai| w2| 99.80000305175781|
+|1970-01-01T08:00:06.000+08:00| Beijing| w1| 103.5|
++-----------------------------+--------+--------+------------------+
+```
+
+Comparing to the pure tag aggregations, this kind of aggregation will divide the data according to the time window specification firstly, and do the aggregation query by the multiple tags in each time window secondly.
+The result set will also contain a time column, which have the same meaning with the time column of the result in downsampling aggregation query by time window.
+
+### Limitation of Aggregation by Tags
+
+As this feature is still under development, some queries have not been completed yet and will be supported in the future.
+
+> 1. Temporarily not support `HAVING` clause to filter the results.
+> 2. Temporarily not support ordering by tag values.
+> 3. Temporarily not support `LIMIT`，`OFFSET`，`SLIMIT`，`SOFFSET`.
+> 4. Temporarily not support `ALIGN BY DEVICE`.
+> 5. Temporarily not support expressions as aggregation function parameter，e.g. `count(s+1)`.
+> 6. Not support the value filter, which stands the same with the `GROUP BY LEVEL` query.
+
## Aggregate result filtering
If you want to filter the results of aggregate queries,
diff --git a/docs/zh/UserGuide/Query-Data/Aggregate-Query.md b/docs/zh/UserGuide/Query-Data/Aggregate-Query.md
index 12e34e0e16..b7b5c285b5 100644
--- a/docs/zh/UserGuide/Query-Data/Aggregate-Query.md
+++ b/docs/zh/UserGuide/Query-Data/Aggregate-Query.md
@@ -448,9 +448,175 @@ Total line number = 7
It costs 0.004s
```
+## 标签聚合查询
+
+IoTDB 还支持通过 `GROUP BY TAGS` 语句根据时间序列中定义的标签的键值做聚合查询。
+
+我们先在 IoTDB 中写入如下示例数据，稍后会以这些数据为例介绍标签聚合查询。
+
+这些是某工厂 `factory1` 在多个城市的多个车间的设备温度数据， 时间范围为 [1000, 10000)。
+
+时间序列路径中的设备一级是设备唯一标识。城市信息 `city` 和车间信息 `workshop` 则被建模在该设备时间序列的标签中。
+其中，设备 `d1`、`d2` 在 `Beijing` 的 `w1` 车间， `d3`、`d4` 在 `Beijing` 的 `w2` 车间，`d5`、`d6` 在 `Shanghai` 的 `w1` 车间，`d7` 在 `Shanghai` 的 `w2` 车间。
+`d8` 和 `d9` 设备目前处于调试阶段，还未被分配到具体的城市和车间，所以其相应的标签值为空值。
+
+```SQL
+set storage group to root.factory1;
+create timeseries root.factory1.d1.temperature with datatype=FLOAT tags(city=Beijing, workshop=w1);
+create timeseries root.factory1.d2.temperature with datatype=FLOAT tags(city=Beijing, workshop=w1);
+create timeseries root.factory1.d3.temperature with datatype=FLOAT tags(city=Beijing, workshop=w2);
+create timeseries root.factory1.d4.temperature with datatype=FLOAT tags(city=Beijing, workshop=w2);
+create timeseries root.factory1.d5.temperature with datatype=FLOAT tags(city=Shanghai, workshop=w1);
+create timeseries root.factory1.d6.temperature with datatype=FLOAT tags(city=Shanghai, workshop=w1);
+create timeseries root.factory1.d7.temperature with datatype=FLOAT tags(city=Shanghai, workshop=w2);
+create timeseries root.factory1.d8.temperature with datatype=FLOAT;
+create timeseries root.factory1.d9.temperature with datatype=FLOAT;
+
+insert into root.factory1.d1(time, temperature) values(1000, 104.0);
+insert into root.factory1.d1(time, temperature) values(3000, 104.2);
+insert into root.factory1.d1(time, temperature) values(5000, 103.3);
+insert into root.factory1.d1(time, temperature) values(7000, 104.1);
+
+insert into root.factory1.d2(time, temperature) values(1000, 104.4);
+insert into root.factory1.d2(time, temperature) values(3000, 103.7);
+insert into root.factory1.d2(time, temperature) values(5000, 103.3);
+insert into root.factory1.d2(time, temperature) values(7000, 102.9);
+
+insert into root.factory1.d3(time, temperature) values(1000, 103.9);
+insert into root.factory1.d3(time, temperature) values(3000, 103.8);
+insert into root.factory1.d3(time, temperature) values(5000, 102.7);
+insert into root.factory1.d3(time, temperature) values(7000, 106.9);
+
+insert into root.factory1.d4(time, temperature) values(1000, 103.9);
+insert into root.factory1.d4(time, temperature) values(5000, 102.7);
+insert into root.factory1.d4(time, temperature) values(7000, 106.9);
+
+insert into root.factory1.d5(time, temperature) values(1000, 112.9);
+insert into root.factory1.d5(time, temperature) values(7000, 113.0);
+
+insert into root.factory1.d6(time, temperature) values(1000, 113.9);
+insert into root.factory1.d6(time, temperature) values(3000, 113.3);
+insert into root.factory1.d6(time, temperature) values(5000, 112.7);
+insert into root.factory1.d6(time, temperature) values(7000, 112.3);
+
+insert into root.factory1.d7(time, temperature) values(1000, 101.2);
+insert into root.factory1.d7(time, temperature) values(3000, 99.3);
+insert into root.factory1.d7(time, temperature) values(5000, 100.1);
+insert into root.factory1.d7(time, temperature) values(7000, 99.8);
+
+insert into root.factory1.d8(time, temperature) values(1000, 50.0);
+insert into root.factory1.d8(time, temperature) values(3000, 52.1);
+insert into root.factory1.d8(time, temperature) values(5000, 50.1);
+insert into root.factory1.d8(time, temperature) values(7000, 50.5);
+
+insert into root.factory1.d9(time, temperature) values(1000, 50.3);
+insert into root.factory1.d9(time, temperature) values(3000, 52.1);
+```
+
+### 单标签聚合查询
+
+用户想统统计该工厂每个地区的设备的温度的平均值，可以使用如下查询语句
+
+```SQL
+SELECT AVG(temperature) FROM root.sg.** GROUP BY TAGS city;
+```
+
+该查询会将具有同一个 `city` 标签值的时间序列的所有满足查询条件的点做平均值计算，计算结果如下
+
+```
++--------+------------------+
+| city| avg(temperature)|
++--------+------------------+
+| Beijing|104.04666697184244|
+|Shanghai|107.85000076293946|
+| NULL| 50.84999910990397|
++--------+------------------+
+Total line number = 3
+It costs 0.231s
+```
+
+从结果集中可以看到，和时间区间聚合、按层次聚合相比，标签聚合的查询结果的不同点是：
+1. 标签聚合查询的聚合结果不会再做去星号展开，而是将多个时间序列的数据作为一个整体进行聚合计算。
+2. 标签聚合查询除了输出聚合结果列，还会输出聚合标签的键值列。该列的列名为聚合指定的标签键，列的值则为所有查询的时间序列中出现的该标签的值。
+如果某些时间序列未设置该标签，则在键值列中有一行单独的 `NULL` ，代表未设置标签的所有时间序列数据的聚合结果。
+
+### 多标签聚合查询
+
+除了基本的单标签聚合查询外，还可以按顺序指定多个标签进行聚合计算。
+
+例如，用户想统计每个城市的每个车间内设备的平均温度。但因为各个城市的车间名称有可能相同，所以不能直接按照 `workshop` 做标签聚合。必须要先按照城市，再按照车间处理。
+
+SQL 语句如下
+
+```SQL
+select avg(temperature) FROM root.factory1.** GROUP BY TAGS city, workshop;
+```
+
+查询结果如下
+
+```
++--------+--------+------------------+
+| city|workshop| avg(temperature)|
++--------+--------+------------------+
+| NULL| NULL| 50.84999910990397|
+|Shanghai| w1|113.01666768391927|
+| Beijing| w2| 104.4000004359654|
+|Shanghai| w2|100.10000038146973|
+| Beijing| w1|103.73750019073486|
++--------+--------+------------------+
+Total line number = 5
+It costs 0.027s
+```
+
+从结果集中可以看到，和单标签聚合相比，多标签聚合的查询结果会根据指定的标签顺序，输出相应标签的键值列。
+
+### 基于时间区间的标签聚合查询
+
+按照时间区间聚合是时序数据库中最常用的查询需求之一。IoTDB 在基于时间区间的聚合基础上，支持进一步按照标签进行聚合查询。
+
+例如，用户想统计时间 `[1000, 10000)` 范围内，每个城市每个车间中的设备每 5 秒内的平均温度。
+
+SQL 语句如下
+
+```SQL
+select avg(temperature) from root.factory1.** group by ([1000, 10000), 5s), tags city, workshop;
+```
+
+查询结果如下
+
+```
++-----------------------------+--------+--------+------------------+
+| Time| city|workshop| avg(temperature)|
++-----------------------------+--------+--------+------------------+
+|1970-01-01T08:00:01.000+08:00| NULL| NULL| 50.91999893188476|
+|1970-01-01T08:00:01.000+08:00|Shanghai| w1|113.20000076293945|
+|1970-01-01T08:00:01.000+08:00| Beijing| w2| 103.4|
+|1970-01-01T08:00:01.000+08:00|Shanghai| w2| 100.1999994913737|
+|1970-01-01T08:00:01.000+08:00| Beijing| w1|103.81666692097981|
+|1970-01-01T08:00:06.000+08:00| NULL| NULL| 50.5|
+|1970-01-01T08:00:06.000+08:00|Shanghai| w1| 112.6500015258789|
+|1970-01-01T08:00:06.000+08:00| Beijing| w2| 106.9000015258789|
+|1970-01-01T08:00:06.000+08:00|Shanghai| w2| 99.80000305175781|
+|1970-01-01T08:00:06.000+08:00| Beijing| w1| 103.5|
++-----------------------------+--------+--------+------------------+
+```
+
+和标签聚合相比，基于时间区间的标签聚合的查询会首先按照时间区间划定聚合范围，在时间区间内部再根据指定的标签顺序，进行相应数据的聚合计算。在输出的结果集中，会包含一列时间列，该时间列值的含义和时间区间聚合查询的相同。
+
+### 标签聚合查询的限制
+
+由于标签聚合功能仍然处于开发阶段，目前有如下未实现功能。
+
+> 1. 暂不支持 `HAVING` 子句过滤查询结果。
+> 2. 暂不支持结果按照标签值排序。
+> 3. 暂不支持 `LIMIT`，`OFFSET`，`SLIMIT`，`SOFFSET`。
+> 4. 暂不支持 `ALIGN BY DEVICE`。
+> 5. 暂不支持聚合函数内部包含表达式，例如 `count(s+1)`。
+> 6. 不支持值过滤条件聚合，和分成聚合查询行为保持一致。
+
## 聚合结果过滤
-如果想对聚合查询的结果进行过滤，可以在`GROUP BY`子句之后使用`HAVING`子句
+如果想对聚合查询的结果进行过滤，可以在 `GROUP BY` 子句之后使用 `HAVING` 子句
> 注意：
>
diff --git a/integration-test/src/test/java/org/apache/iotdb/db/it/aggregation/IoTDBTagAggregationIT.java b/integration-test/src/test/java/org/apache/iotdb/db/it/aggregation/IoTDBTagAggregationIT.java
new file mode 100644
index 0000000000..c5461331a4
--- /dev/null
+++ b/integration-test/src/test/java/org/apache/iotdb/db/it/aggregation/IoTDBTagAggregationIT.java
@@ -0,0 +1,445 @@
+/*
+ * 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.it.aggregation;
+
+import org.apache.iotdb.it.env.EnvFactory;
+import org.apache.iotdb.itbase.category.ClusterIT;
+
+import org.junit.AfterClass;
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Ignore;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+import java.sql.Connection;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+import static org.junit.Assert.fail;
+
+@Category({ClusterIT.class})
+public class IoTDBTagAggregationIT {
+ private static final String[] DATASET =
+ new String[] {
+ "create timeseries root.sg.d1.t with datatype=FLOAT tags(k1=k1v1, k2=k2v1, k3=k3v1);",
+ "create timeseries root.sg.d2.t with datatype=FLOAT tags(k1=k1v1, k2=k2v2);",
+ "create timeseries root.sg.d3.t with datatype=FLOAT tags(k1=k1v2, k2=k2v1);",
+ "create timeseries root.sg.d4.t with datatype=FLOAT tags(k1=k1v2, k2=k2v2);",
+ "create timeseries root.sg.d5.t with datatype=FLOAT tags(k1=k1v1);",
+ "create timeseries root.sg.d6.t with datatype=FLOAT tags(k2=k2v1);",
+ "create timeseries root.sg.d7.t with datatype=FLOAT;",
+ "create timeseries root.sg2.d8.t with datatype=TEXT tags(k3=k3v1);",
+ "insert into root.sg.d1(time, t) values(1, 1.1);",
+ "insert into root.sg.d2(time, t) values(1, 1.2);",
+ "insert into root.sg.d3(time, t) values(1, 1.3);",
+ "insert into root.sg.d4(time, t) values(1, 1.4);",
+ "insert into root.sg.d5(time, t) values(1, 1.5);",
+ "insert into root.sg.d6(time, t) values(1, 1.6);",
+ "insert into root.sg.d7(time, t) values(1, 1.7);",
+ "insert into root.sg2.d8(time, t) values(1, 'abc');",
+ "insert into root.sg.d1(time, t) values(10, 2.1);",
+ "insert into root.sg.d2(time, t) values(10, 3.2);",
+ "insert into root.sg.d3(time, t) values(10, 4.3);",
+ "insert into root.sg.d4(time, t) values(10, 5.4);",
+ "insert into root.sg.d5(time, t) values(10, 6.5);",
+ "insert into root.sg.d6(time, t) values(10, 7.6);",
+ "insert into root.sg.d7(time, t) values(10, 8.7);"
+ };
+
+ private static final double DELTA = 0.001D;
+
+ @BeforeClass
+ public static void setUp() throws Exception {
+ EnvFactory.getEnv().initBeforeClass();
+ try (Connection connection = EnvFactory.getEnv().getConnection();
+ Statement statement = connection.createStatement()) {
+ for (String sql : DATASET) {
+ statement.execute(sql);
+ }
+ } catch (Exception e) {
+ e.printStackTrace();
+ throw e;
+ }
+ }
+
+ @AfterClass
+ public static void tearDown() throws Exception {
+ EnvFactory.getEnv().cleanAfterClass();
+ }
+
+ @Test
+ public void testAggregateFunctions() {
+ String query =
+ "SELECT COUNT(t), AVG(t), MAX_TIME(t), MIN_TIME(t), MAX_VALUE(t), MIN_VALUE(t), EXTREME(t) FROM root.sg.** GROUP BY TAGS k1";
+ // Expected result set:
+ // +----+--------+------------------+-----------+-----------+------------+------------+----------+
+ // | k1|count(t)|
+ // avg(t)|max_time(t)|min_time(t)|max_value(t)|min_value(t)|extreme(t)|
+ // +----+--------+------------------+-----------+-----------+------------+------------+----------+
+ // |k1v2| 4|3.1000000536441803| 10| 1| 5.4| 1.3|
+ // 5.4|
+ // |k1v1| 6| 2.600000003973643| 10| 1| 6.5| 1.1|
+ // 6.5|
+ // |NULL| 4| 4.89999994635582| 10| 1| 8.7| 1.6|
+ // 8.7|
+ // +----+--------+------------------+-----------+-----------+------------+------------+----------+
+ try (Connection connection = EnvFactory.getEnv().getConnection();
+ Statement statement = connection.createStatement()) {
+ try (ResultSet resultSet = statement.executeQuery(query)) {
+ Assert.assertEquals(8, resultSet.getMetaData().getColumnCount());
+ Assert.assertTrue(resultSet.next());
+ Assert.assertEquals("k1v2", resultSet.getString("k1"));
+ Assert.assertEquals(4L, resultSet.getLong("count(t)"));
+ Assert.assertEquals(3.1D, resultSet.getDouble("avg(t)"), DELTA);
+ Assert.assertEquals(10L, resultSet.getLong("max_time(t)"));
+ Assert.assertEquals(1L, resultSet.getLong("min_time(t)"));
+ Assert.assertEquals(5.4F, resultSet.getFloat("max_value(t)"), DELTA);
+ Assert.assertEquals(1.3F, resultSet.getFloat("min_value(t)"), DELTA);
+ Assert.assertEquals(5.4F, resultSet.getFloat("extreme(t)"), DELTA);
+ Assert.assertTrue(resultSet.next());
+ Assert.assertEquals("k1v1", resultSet.getString(1));
+ Assert.assertEquals(6L, resultSet.getLong(2));
+ Assert.assertEquals(2.6D, resultSet.getDouble(3), DELTA);
+ Assert.assertEquals(10L, resultSet.getLong(4));
+ Assert.assertEquals(1L, resultSet.getLong(5));
+ Assert.assertEquals(6.5F, resultSet.getFloat(6), DELTA);
+ Assert.assertEquals(1.1F, resultSet.getFloat(7), DELTA);
+ Assert.assertEquals(6.5F, resultSet.getFloat(8), DELTA);
+ Assert.assertTrue(resultSet.next());
+ Assert.assertEquals("NULL", resultSet.getString(1));
+ Assert.assertEquals(4L, resultSet.getLong(2));
+ Assert.assertEquals(4.9D, resultSet.getDouble(3), DELTA);
+ Assert.assertEquals(10L, resultSet.getLong(4));
+ Assert.assertEquals(1L, resultSet.getLong(5));
+ Assert.assertEquals(8.7D, resultSet.getFloat(6), DELTA);
+ Assert.assertEquals(1.6D, resultSet.getFloat(7), DELTA);
+ Assert.assertEquals(8.7D, resultSet.getFloat(8), DELTA);
+ Assert.assertFalse(resultSet.next());
+ }
+ } catch (SQLException e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ @Test
+ @Ignore
+ public void testAggregateFunctionsWithNestedExpression() {
+ String query = "SELECT COUNT(t + 1), AVG(t + 1) FROM root.sg.** GROUP BY TAGS k1";
+ // Expected result set:
+ // +----+------------+------------------+
+ // | k1|count(t + 1)| avg(t + 1)|
+ // +----+------------+------------------+
+ // |k1v2| 4|3.1000000536441803|
+ // |k1v1| 6| 3.600000003973643|
+ // |NULL| 4| 5.89999994635582|
+ // +----+------------+------------------+
+ try (Connection connection = EnvFactory.getEnv().getConnection();
+ Statement statement = connection.createStatement()) {
+ try (ResultSet resultSet = statement.executeQuery(query)) {
+ Assert.assertEquals(8, resultSet.getMetaData().getColumnCount());
+ Assert.assertTrue(resultSet.next());
+ Assert.assertEquals("k1v2", resultSet.getString("k1"));
+ Assert.assertEquals(4L, resultSet.getLong("count(t)"));
+ Assert.assertEquals(3.1D, resultSet.getDouble("avg(t)"), DELTA);
+ Assert.assertEquals(10L, resultSet.getLong("max_time(t)"));
+ Assert.assertEquals(1L, resultSet.getLong("min_time(t)"));
+ Assert.assertEquals(5.4F, resultSet.getFloat("max_value(t)"), DELTA);
+ Assert.assertEquals(1.3F, resultSet.getFloat("min_value(t)"), DELTA);
+ Assert.assertEquals(5.4F, resultSet.getFloat("extreme(t)"), DELTA);
+ Assert.assertTrue(resultSet.next());
+ Assert.assertEquals("k1v1", resultSet.getString(1));
+ Assert.assertEquals(6L, resultSet.getLong(2));
+ Assert.assertEquals(2.6D, resultSet.getDouble(3), DELTA);
+ Assert.assertEquals(10L, resultSet.getLong(4));
+ Assert.assertEquals(1L, resultSet.getLong(5));
+ Assert.assertEquals(6.5F, resultSet.getFloat(6), DELTA);
+ Assert.assertEquals(1.1F, resultSet.getFloat(7), DELTA);
+ Assert.assertEquals(6.5F, resultSet.getFloat(8), DELTA);
+ Assert.assertTrue(resultSet.next());
+ Assert.assertEquals("NULL", resultSet.getString(1));
+ Assert.assertEquals(4L, resultSet.getLong(2));
+ Assert.assertEquals(4.9D, resultSet.getDouble(3), DELTA);
+ Assert.assertEquals(10L, resultSet.getLong(4));
+ Assert.assertEquals(1L, resultSet.getLong(5));
+ Assert.assertEquals(8.7D, resultSet.getFloat(6), DELTA);
+ Assert.assertEquals(1.6D, resultSet.getFloat(7), DELTA);
+ Assert.assertEquals(8.7D, resultSet.getFloat(8), DELTA);
+ Assert.assertFalse(resultSet.next());
+ }
+ } catch (SQLException e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ @Test
+ @Ignore // TODO: support having in later commits
+ public void testAggregateFunctionsWithHaving() {
+ String query =
+ "SELECT COUNT(t), AVG(t), MAX_TIME(t), MIN_TIME(t), MAX_VALUE(t), MIN_VALUE(t), EXTREME(t) FROM root.sg.** GROUP BY TAGS k1 HAVING avg(t) > 3";
+ // Expected result set:
+ // +----+--------+------------------+-----------+-----------+------------+------------+----------+
+ // | k1|count(t)|
+ // avg(t)|max_time(t)|min_time(t)|max_value(t)|min_value(t)|extreme(t)|
+ // +----+--------+------------------+-----------+-----------+------------+------------+----------+
+ // |k1v2| 4|3.1000000536441803| 10| 1| 5.4| 1.3|
+ // 5.4|
+ // |NULL| 4| 4.89999994635582| 10| 1| 8.7| 1.6|
+ // 8.7|
+ // +----+--------+------------------+-----------+-----------+------------+------------+----------+
+ try (Connection connection = EnvFactory.getEnv().getConnection();
+ Statement statement = connection.createStatement()) {
+ try (ResultSet resultSet = statement.executeQuery(query)) {
+ Assert.assertEquals(8, resultSet.getMetaData().getColumnCount());
+ Assert.assertTrue(resultSet.next());
+ Assert.assertEquals("k1v2", resultSet.getString("k1"));
+ Assert.assertEquals(4L, resultSet.getLong("count(t)"));
+ Assert.assertEquals(3.1D, resultSet.getDouble("avg(t)"), DELTA);
+ Assert.assertEquals(10L, resultSet.getLong("max_time(t)"));
+ Assert.assertEquals(1L, resultSet.getLong("min_time(t)"));
+ Assert.assertEquals(5.4F, resultSet.getFloat("max_value(t)"), DELTA);
+ Assert.assertEquals(1.3F, resultSet.getFloat("min_value(t)"), DELTA);
+ Assert.assertEquals(5.4F, resultSet.getFloat("extreme(t)"), DELTA);
+ Assert.assertTrue(resultSet.next());
+ Assert.assertEquals("NULL", resultSet.getString(1));
+ Assert.assertEquals(4L, resultSet.getLong(2));
+ Assert.assertEquals(4.9D, resultSet.getDouble(3), DELTA);
+ Assert.assertEquals(10L, resultSet.getLong(4));
+ Assert.assertEquals(1L, resultSet.getLong(5));
+ Assert.assertEquals(8.7D, resultSet.getFloat(6), DELTA);
+ Assert.assertEquals(1.6D, resultSet.getFloat(7), DELTA);
+ Assert.assertEquals(8.7D, resultSet.getFloat(8), DELTA);
+ Assert.assertFalse(resultSet.next());
+ }
+ } catch (SQLException e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ @Test
+ public void testMultipleAggregationKeys() {
+ String query = "SELECT COUNT(t) FROM root.sg.** GROUP BY TAGS k1, k2";
+ // Expected result set:
+ // +----+----+--------+
+ // | k1| k2|count(t)|
+ // +----+----+--------+
+ // |NULL|NULL| 2|
+ // |NULL|k2v1| 2|
+ // |k1v1|NULL| 2|
+ // |k1v2|k2v1| 2|
+ // |k1v1|k2v2| 2|
+ // |k1v1|k2v1| 2|
+ // |k1v2|k2v2| 2|
+ // +----+----+--------+
+ try (Connection connection = EnvFactory.getEnv().getConnection();
+ Statement statement = connection.createStatement()) {
+ try (ResultSet resultSet = statement.executeQuery(query)) {
+ Assert.assertEquals(3, resultSet.getMetaData().getColumnCount());
+ Set<List<String>> groups = new HashSet<>();
+ for (int i = 0; i < 7; i++) {
+ Assert.assertTrue(resultSet.next());
+ List<String> tagValues = new ArrayList<>(2);
+ tagValues.add(resultSet.getString("k1"));
+ tagValues.add(resultSet.getString("k2"));
+ groups.add(tagValues);
+ Assert.assertEquals(2L, resultSet.getLong("count(t)"));
+ }
+ Assert.assertFalse(resultSet.next());
+ Assert.assertEquals(7, groups.size());
+ }
+ } catch (SQLException e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ @Test
+ public void testAlongWithTimeAggregation() {
+ String query = "SELECT COUNT(t) from root.sg.** GROUP BY ([0, 20), 10ms), TAGS k1";
+ // Expected result set:
+ // +-----------------------------+----+--------+
+ // | Time| k1|count(t)|
+ // +-----------------------------+----+--------+
+ // |1970-01-01T08:00:00.000+08:00|k1v2| 2|
+ // |1970-01-01T08:00:00.000+08:00|k1v1| 3|
+ // |1970-01-01T08:00:00.000+08:00|NULL| 2|
+ // |1970-01-01T08:00:00.010+08:00|k1v2| 2|
+ // |1970-01-01T08:00:00.010+08:00|k1v1| 3|
+ // |1970-01-01T08:00:00.010+08:00|NULL| 2|
+ // +-----------------------------+----+--------+
+ try (Connection connection = EnvFactory.getEnv().getConnection();
+ Statement statement = connection.createStatement()) {
+ try (ResultSet resultSet = statement.executeQuery(query)) {
+ Assert.assertEquals(3, resultSet.getMetaData().getColumnCount());
+ Set<String> groups = new HashSet<>();
+ for (int i = 0; i < 6; i++) {
+ Assert.assertTrue(resultSet.next());
+ if (i < 3) {
+ Assert.assertEquals(0L, resultSet.getLong("Time"));
+ } else {
+ Assert.assertEquals(10L, resultSet.getLong("Time"));
+ }
+ String tagValue = resultSet.getString("k1");
+ switch (tagValue) {
+ case "k1v1":
+ Assert.assertEquals(3L, resultSet.getLong("count(t)"));
+ break;
+ case "k1v2":
+ Assert.assertEquals(2L, resultSet.getLong("count(t)"));
+ break;
+ case "NULL":
+ Assert.assertEquals(2L, resultSet.getLong("count(t)"));
+ break;
+ default:
+ fail("Unexpected tag value: " + tagValue);
+ }
+ groups.add(tagValue);
+ }
+ Assert.assertEquals(3, groups.size());
+ Assert.assertFalse(resultSet.next());
+ }
+ } catch (SQLException e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ @Test
+ public void testAlongWithTimeAggregationAndOrdering() {
+ String query =
+ "SELECT COUNT(t) from root.sg.** GROUP BY ([0, 20), 10ms), TAGS k1 ORDER BY TIME DESC";
+ // Expected result set:
+ try (Connection connection = EnvFactory.getEnv().getConnection();
+ Statement statement = connection.createStatement()) {
+ try (ResultSet resultSet = statement.executeQuery(query)) {
+ Assert.assertEquals(3, resultSet.getMetaData().getColumnCount());
+ Set<String> groups = new HashSet<>();
+ for (int i = 0; i < 6; i++) {
+ Assert.assertTrue(resultSet.next());
+ if (i < 3) {
+ Assert.assertEquals(10L, resultSet.getLong("Time"));
+ } else {
+ Assert.assertEquals(0L, resultSet.getLong("Time"));
+ }
+ String tagValue = resultSet.getString("k1");
+ switch (tagValue) {
+ case "k1v1":
+ Assert.assertEquals(3L, resultSet.getLong("count(t)"));
+ break;
+ case "k1v2":
+ Assert.assertEquals(2L, resultSet.getLong("count(t)"));
+ break;
+ case "NULL":
+ Assert.assertEquals(2L, resultSet.getLong("count(t)"));
+ break;
+ default:
+ fail("Unexpected tag value: " + tagValue);
+ }
+ groups.add(tagValue);
+ }
+ Assert.assertEquals(3, groups.size());
+ Assert.assertFalse(resultSet.next());
+ }
+ } catch (SQLException e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ @Test
+ public void testAlongWithTimeFiltering() {
+ String query = "SELECT COUNT(t) FROM root.sg.** WHERE time > 1 GROUP BY TAGS k1";
+ // Expected result set:
+ // +----+--------+
+ // | k1|count(t)|
+ // +----+--------+
+ // |k1v2| 2|
+ // |k1v1| 3|
+ // |NULL| 2|
+ // +----+--------+
+ try (Connection connection = EnvFactory.getEnv().getConnection();
+ Statement statement = connection.createStatement()) {
+ try (ResultSet resultSet = statement.executeQuery(query)) {
+ Assert.assertEquals(2, resultSet.getMetaData().getColumnCount());
+ Set<String> groups = new HashSet<>();
+ for (int i = 0; i < 3; i++) {
+ Assert.assertTrue(resultSet.next());
+ String tagValue = resultSet.getString("k1");
+ switch (tagValue) {
+ case "k1v1":
+ Assert.assertEquals(3L, resultSet.getLong("count(t)"));
+ break;
+ case "k1v2":
+ Assert.assertEquals(2L, resultSet.getLong("count(t)"));
+ break;
+ case "NULL":
+ Assert.assertEquals(2L, resultSet.getLong("count(t)"));
+ break;
+ default:
+ fail("Unexpected tag value: " + tagValue);
+ }
+ groups.add(tagValue);
+ }
+ Assert.assertEquals(3, groups.size());
+ Assert.assertFalse(resultSet.next());
+ }
+ } catch (SQLException e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ @Test
+ public void testIncompatibleMixedDataTypes() {
+ String query = "SELECT AVG(t) FROM root.** GROUP BY TAGS k3";
+ // AVG() with numeric and text timeseries, an exception will be thrown
+ try (Connection connection = EnvFactory.getEnv().getConnection();
+ Statement statement = connection.createStatement()) {
+ try (ResultSet ignored = statement.executeQuery(query)) {
+ Assert.fail();
+ }
+ } catch (SQLException e) {
+ Assert.assertTrue(e.getMessage().contains("only support numeric data types"));
+ }
+ }
+
+ @Test
+ public void testWithValueFilters() {
+ String query = "SELECT AVG(t) FROM root.sg.** WHERE t > 1.5 GROUP BY TAGS k1";
+ // Value filter is not supported yet
+ try (Connection connection = EnvFactory.getEnv().getConnection();
+ Statement statement = connection.createStatement()) {
+ try (ResultSet ignored = statement.executeQuery(query)) {
+ Assert.fail();
+ }
+ } catch (SQLException e) {
+ Assert.assertTrue(
+ e.getMessage().contains("Only time filters are supported in GROUP BY TAGS query"));
+ }
+ }
+}
diff --git a/schema-engine-rocksdb/src/main/java/org/apache/iotdb/db/metadata/schemaregion/rocksdb/RSchemaRegion.java b/schema-engine-rocksdb/src/main/java/org/apache/iotdb/db/metadata/schemaregion/rocksdb/RSchemaRegion.java
index f82465708f..94a7a19931 100644
--- a/schema-engine-rocksdb/src/main/java/org/apache/iotdb/db/metadata/schemaregion/rocksdb/RSchemaRegion.java
+++ b/schema-engine-rocksdb/src/main/java/org/apache/iotdb/db/metadata/schemaregion/rocksdb/RSchemaRegion.java
@@ -117,6 +117,7 @@ import java.util.concurrent.locks.ReentrantReadWriteLock;
import java.util.function.BiFunction;
import java.util.function.Function;
import java.util.stream.Collectors;
+import java.util.stream.Stream;
import static org.apache.iotdb.db.metadata.schemaregion.rocksdb.RSchemaConstants.ALL_NODE_TYPE_ARRAY;
import static org.apache.iotdb.db.metadata.schemaregion.rocksdb.RSchemaConstants.DEFAULT_ALIGNED_ENTITY_VALUE;
@@ -641,29 +642,28 @@ public class RSchemaRegion implements ISchemaRegion {
}
Set<IMNode> tempSet = ConcurrentHashMap.newKeySet();
- parentNeedsToCheck
- .parallelStream()
- .forEach(
- currentNode -> {
- if (!currentNode.isStorageGroup()) {
- PartialPath parentPath = currentNode.getPartialPath();
- int level = parentPath.getNodeLength();
- int end = parentPath.getNodeLength() - 1;
- if (!readWriteHandler.existAnySiblings(
- RSchemaUtils.getLevelPathPrefix(parentPath.getNodes(), end, level))) {
- try {
- readWriteHandler.deleteNode(
- parentPath.getNodes(), RSchemaUtils.typeOfMNode(currentNode));
- IMNode parentNode = currentNode.getParent();
- if (!parentNode.isStorageGroup()) {
- tempSet.add(currentNode.getParent());
- }
- } catch (Exception e) {
- logger.warn("delete {} fail.", parentPath.getFullPath(), e);
- }
+ Stream<IMNode> parentStream = parentNeedsToCheck.parallelStream();
+ parentStream.forEach(
+ currentNode -> {
+ if (!currentNode.isStorageGroup()) {
+ PartialPath parentPath = currentNode.getPartialPath();
+ int level = parentPath.getNodeLength();
+ int end = parentPath.getNodeLength() - 1;
+ if (!readWriteHandler.existAnySiblings(
+ RSchemaUtils.getLevelPathPrefix(parentPath.getNodes(), end, level))) {
+ try {
+ readWriteHandler.deleteNode(
+ parentPath.getNodes(), RSchemaUtils.typeOfMNode(currentNode));
+ IMNode parentNode = currentNode.getParent();
+ if (!parentNode.isStorageGroup()) {
+ tempSet.add(currentNode.getParent());
}
+ } catch (Exception e) {
+ logger.warn("delete {} fail.", parentPath.getFullPath(), e);
}
- });
+ }
+ }
+ });
parentNeedsToCheck.clear();
parentNeedsToCheck.addAll(tempSet);
}
@@ -740,35 +740,33 @@ public class RSchemaRegion implements ISchemaRegion {
byte[] suffixToMatch =
RSchemaUtils.getSuffixOfLevelPath(
ArrayUtils.subarray(nodes, firstNonWildcardIndex, nextFirstWildcardIndex), level);
-
- scanKeys
- .parallelStream()
- .forEach(
- prefixNodes -> {
- String levelPrefix =
- RSchemaUtils.getLevelPathPrefix(prefixNodes, prefixNodes.length - 1, level);
- Arrays.stream(nodeType)
- .parallel()
- .forEach(
- x -> {
- byte[] startKey = RSchemaUtils.toRocksDBKey(levelPrefix, x);
- RocksIterator iterator = readWriteHandler.iterator(null);
- iterator.seek(startKey);
- while (iterator.isValid()) {
- if (!RSchemaUtils.prefixMatch(iterator.key(), startKey)) {
- break;
- }
- if (RSchemaUtils.suffixMatch(iterator.key(), suffixToMatch)) {
- if (lastIteration) {
- function.apply(iterator.key(), iterator.value());
- } else {
- tempNodes.add(RSchemaUtils.toMetaNodes(iterator.key()));
- }
- }
- iterator.next();
+ Stream<String[]> scanKeysStream = scanKeys.parallelStream();
+ scanKeysStream.forEach(
+ prefixNodes -> {
+ String levelPrefix =
+ RSchemaUtils.getLevelPathPrefix(prefixNodes, prefixNodes.length - 1, level);
+ Arrays.stream(nodeType)
+ .parallel()
+ .forEach(
+ x -> {
+ byte[] startKey = RSchemaUtils.toRocksDBKey(levelPrefix, x);
+ RocksIterator iterator = readWriteHandler.iterator(null);
+ iterator.seek(startKey);
+ while (iterator.isValid()) {
+ if (!RSchemaUtils.prefixMatch(iterator.key(), startKey)) {
+ break;
+ }
+ if (RSchemaUtils.suffixMatch(iterator.key(), suffixToMatch)) {
+ if (lastIteration) {
+ function.apply(iterator.key(), iterator.value());
+ } else {
+ tempNodes.add(RSchemaUtils.toMetaNodes(iterator.key()));
}
- });
- });
+ }
+ iterator.next();
+ }
+ });
+ });
scanKeys.clear();
scanKeys.addAll(tempNodes);
tempNodes.clear();
@@ -1176,8 +1174,13 @@ public class RSchemaRegion implements ISchemaRegion {
}
@Override
- public List<MeasurementPath> getMeasurementPaths(PartialPath pathPattern, boolean isPrefixMath)
- throws MetadataException {
+ public List<MeasurementPath> getMeasurementPaths(
+ PartialPath pathPattern, boolean isPrefixMath, boolean withTags) throws MetadataException {
+ if (withTags) {
+ Map<MeasurementPath, Pair<Map<String, String>, Map<String, String>>> results =
+ getMatchedMeasurementPathWithTags(pathPattern.getNodes());
+ return new ArrayList<>(results.keySet());
+ }
List<MeasurementPath> allResult = Collections.synchronizedList(new ArrayList<>());
BiFunction<byte[], byte[], Boolean> function =
(a, b) -> {
@@ -1195,15 +1198,16 @@ public class RSchemaRegion implements ISchemaRegion {
@Override
public Pair<List<MeasurementPath>, Integer> getMeasurementPathsWithAlias(
- PartialPath pathPattern, int limit, int offset, boolean isPrefixMatch)
+ PartialPath pathPattern, int limit, int offset, boolean isPrefixMatch, boolean withTags)
throws MetadataException {
// todo page query
- return new Pair<>(getMeasurementPaths(pathPattern, false), offset + limit);
+ return new Pair<>(getMeasurementPaths(pathPattern, false, withTags), offset + limit);
}
@Override
public List<MeasurementPath> fetchSchema(
- PartialPath pathPattern, Map<Integer, Template> templateMap) throws MetadataException {
+ PartialPath pathPattern, Map<Integer, Template> templateMap, boolean withTags)
+ throws MetadataException {
throw new UnsupportedOperationException();
}
@@ -1249,6 +1253,7 @@ public class RSchemaRegion implements ISchemaRegion {
return new Pair<>(res, 1);
}
+ @SuppressWarnings("unchecked")
private Map<MeasurementPath, Pair<Map<String, String>, Map<String, String>>>
getMatchedMeasurementPathWithTags(String[] nodes) throws IllegalPathException {
Map<MeasurementPath, Pair<Map<String, String>, Map<String, String>>> allResult =
@@ -1267,10 +1272,12 @@ public class RSchemaRegion implements ISchemaRegion {
if (!(attributes instanceof Map)) {
attributes = Collections.emptyMap();
}
- @SuppressWarnings("unchecked")
+ Map<String, String> tagMap = (Map<String, String>) tag;
+ Map<String, String> attributesMap = (Map<String, String>) attributes;
Pair<Map<String, String>, Map<String, String>> tagsAndAttributes =
- new Pair<>((Map<String, String>) tag, (Map<String, String>) attributes);
+ new Pair<>(tagMap, attributesMap);
allResult.put(measurementPath, tagsAndAttributes);
+ measurementPath.setTagMap(tagMap);
return true;
};
traverseOutcomeBasins(nodes, MAX_PATH_DEPTH, function, new Character[] {NODE_TYPE_MEASUREMENT});
diff --git a/server/src/main/java/org/apache/iotdb/db/metadata/LocalSchemaProcessor.java b/server/src/main/java/org/apache/iotdb/db/metadata/LocalSchemaProcessor.java
index fafd13ddd9..c4cc1958f5 100644
--- a/server/src/main/java/org/apache/iotdb/db/metadata/LocalSchemaProcessor.java
+++ b/server/src/main/java/org/apache/iotdb/db/metadata/LocalSchemaProcessor.java
@@ -798,7 +798,7 @@ public class LocalSchemaProcessor {
*/
public List<MeasurementPath> getMeasurementPaths(PartialPath pathPattern, boolean isPrefixMatch)
throws MetadataException {
- return getMeasurementPathsWithAlias(pathPattern, 0, 0, isPrefixMatch).left;
+ return getMeasurementPathsWithAlias(pathPattern, 0, 0, isPrefixMatch, false).left;
}
/**
@@ -821,7 +821,7 @@ public class LocalSchemaProcessor {
* @param isPrefixMatch if true, the path pattern is used to match prefix path
*/
public Pair<List<MeasurementPath>, Integer> getMeasurementPathsWithAlias(
- PartialPath pathPattern, int limit, int offset, boolean isPrefixMatch)
+ PartialPath pathPattern, int limit, int offset, boolean isPrefixMatch, boolean withTags)
throws MetadataException {
List<MeasurementPath> measurementPaths = new LinkedList<>();
Pair<List<MeasurementPath>, Integer> result;
@@ -836,7 +836,7 @@ public class LocalSchemaProcessor {
}
result =
schemaRegion.getMeasurementPathsWithAlias(
- pathPattern, tmpLimit, tmpOffset, isPrefixMatch);
+ pathPattern, tmpLimit, tmpOffset, isPrefixMatch, withTags);
measurementPaths.addAll(result.left);
resultOffset += result.right;
if (limit != 0) {
diff --git a/server/src/main/java/org/apache/iotdb/db/metadata/cache/DataNodeSchemaCache.java b/server/src/main/java/org/apache/iotdb/db/metadata/cache/DataNodeSchemaCache.java
index 6c88cbb870..a1e20f80f3 100644
--- a/server/src/main/java/org/apache/iotdb/db/metadata/cache/DataNodeSchemaCache.java
+++ b/server/src/main/java/org/apache/iotdb/db/metadata/cache/DataNodeSchemaCache.java
@@ -87,6 +87,7 @@ public class DataNodeSchemaCache {
devicePath.concatNode(
schemaCacheEntry.getSchemaEntryId()), // the cached path may be alias path
schemaCacheEntry.getMeasurementSchema(),
+ schemaCacheEntry.getTagMap(),
null,
schemaCacheEntry.isAligned());
}
@@ -99,6 +100,7 @@ public class DataNodeSchemaCache {
SchemaCacheEntry schemaCacheEntry =
new SchemaCacheEntry(
(MeasurementSchema) measurementPath.getMeasurementSchema(),
+ measurementPath.getTagMap(),
measurementPath.isUnderAlignedEntity());
cache.put(new PartialPath(measurementPath.getNodes()), schemaCacheEntry);
}
@@ -146,6 +148,7 @@ public class DataNodeSchemaCache {
entry =
new SchemaCacheEntry(
(MeasurementSchema) measurementPath.getMeasurementSchema(),
+ measurementPath.getTagMap(),
measurementPath.isUnderAlignedEntity());
cache.put(seriesPath, entry);
}
diff --git a/server/src/main/java/org/apache/iotdb/db/metadata/cache/SchemaCacheEntry.java b/server/src/main/java/org/apache/iotdb/db/metadata/cache/SchemaCacheEntry.java
index 03ddf17360..9b0fa929d7 100644
--- a/server/src/main/java/org/apache/iotdb/db/metadata/cache/SchemaCacheEntry.java
+++ b/server/src/main/java/org/apache/iotdb/db/metadata/cache/SchemaCacheEntry.java
@@ -24,17 +24,22 @@ import org.apache.iotdb.db.metadata.lastCache.container.LastCacheContainer;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
+import java.util.Map;
+
public class SchemaCacheEntry {
private final MeasurementSchema measurementSchema;
+ private final Map<String, String> tagMap;
private final boolean isAligned;
private volatile ILastCacheContainer lastCacheContainer = null;
- SchemaCacheEntry(MeasurementSchema measurementSchema, boolean isAligned) {
+ SchemaCacheEntry(
+ MeasurementSchema measurementSchema, Map<String, String> tagMap, boolean isAligned) {
this.measurementSchema = measurementSchema;
this.isAligned = isAligned;
+ this.tagMap = tagMap;
}
public String getSchemaEntryId() {
@@ -45,6 +50,10 @@ public class SchemaCacheEntry {
return measurementSchema;
}
+ public Map<String, String> getTagMap() {
+ return tagMap;
+ }
+
public TSDataType getTsDataType() {
return measurementSchema.getType();
}
diff --git a/server/src/main/java/org/apache/iotdb/db/metadata/mtree/IMTreeBelowSG.java b/server/src/main/java/org/apache/iotdb/db/metadata/mtree/IMTreeBelowSG.java
index f1e2b8cdce..615bb42e95 100644
--- a/server/src/main/java/org/apache/iotdb/db/metadata/mtree/IMTreeBelowSG.java
+++ b/server/src/main/java/org/apache/iotdb/db/metadata/mtree/IMTreeBelowSG.java
@@ -136,12 +136,15 @@ public interface IMTreeBelowSG {
* collected and return.
*
* @param pathPattern a path pattern or a full path, may contain wildcard
+ * @param limit the limit of query result.
+ * @param offset the offset.
* @param isPrefixMatch if true, the path pattern is used to match prefix path
+ * @param withTags whether returns all the tags of each timeseries as well.
* @return Pair.left contains all the satisfied path Pair.right means the current offset or zero
* if we don't set offset.
*/
Pair<List<MeasurementPath>, Integer> getMeasurementPathsWithAlias(
- PartialPath pathPattern, int limit, int offset, boolean isPrefixMatch)
+ PartialPath pathPattern, int limit, int offset, boolean isPrefixMatch, boolean withTags)
throws MetadataException;
/**
diff --git a/server/src/main/java/org/apache/iotdb/db/metadata/mtree/MTreeBelowSGCachedImpl.java b/server/src/main/java/org/apache/iotdb/db/metadata/mtree/MTreeBelowSGCachedImpl.java
index 7247047d7a..6441513b73 100644
--- a/server/src/main/java/org/apache/iotdb/db/metadata/mtree/MTreeBelowSGCachedImpl.java
+++ b/server/src/main/java/org/apache/iotdb/db/metadata/mtree/MTreeBelowSGCachedImpl.java
@@ -51,6 +51,7 @@ import org.apache.iotdb.db.metadata.mtree.traverser.counter.MNodeLevelCounter;
import org.apache.iotdb.db.metadata.mtree.traverser.counter.MeasurementCounter;
import org.apache.iotdb.db.metadata.mtree.traverser.counter.MeasurementGroupByLevelCounter;
import org.apache.iotdb.db.metadata.path.MeasurementPath;
+import org.apache.iotdb.db.metadata.tag.TagManager;
import org.apache.iotdb.db.metadata.template.Template;
import org.apache.iotdb.db.metadata.utils.MetaFormatUtils;
import org.apache.iotdb.db.qp.physical.sys.ShowDevicesPlan;
@@ -114,11 +115,14 @@ public class MTreeBelowSGCachedImpl implements IMTreeBelowSG {
private CachedMTreeStore store;
private volatile IStorageGroupMNode storageGroupMNode;
+ private final TagManager tagManager;
private int levelOfSG;
// region MTree initialization, clear and serialization
- public MTreeBelowSGCachedImpl(IStorageGroupMNode storageGroupMNode, int schemaRegionId)
+ public MTreeBelowSGCachedImpl(
+ IStorageGroupMNode storageGroupMNode, TagManager tagManager, int schemaRegionId)
throws MetadataException, IOException {
+ this.tagManager = tagManager;
PartialPath storageGroup = storageGroupMNode.getPartialPath();
store = new CachedMTreeStore(storageGroup, schemaRegionId);
this.storageGroupMNode = store.getRoot().getAsStorageGroupMNode();
@@ -631,7 +635,7 @@ public class MTreeBelowSGCachedImpl implements IMTreeBelowSG {
@Override
public List<MeasurementPath> getMeasurementPaths(PartialPath pathPattern, boolean isPrefixMatch)
throws MetadataException {
- return getMeasurementPathsWithAlias(pathPattern, 0, 0, isPrefixMatch).left;
+ return getMeasurementPathsWithAlias(pathPattern, 0, 0, isPrefixMatch, false).left;
}
/**
@@ -657,7 +661,7 @@ public class MTreeBelowSGCachedImpl implements IMTreeBelowSG {
*/
@Override
public Pair<List<MeasurementPath>, Integer> getMeasurementPathsWithAlias(
- PartialPath pathPattern, int limit, int offset, boolean isPrefixMatch)
+ PartialPath pathPattern, int limit, int offset, boolean isPrefixMatch, boolean withTags)
throws MetadataException {
List<MeasurementPath> result = new LinkedList<>();
MeasurementCollector<List<PartialPath>> collector =
@@ -670,6 +674,13 @@ public class MTreeBelowSGCachedImpl implements IMTreeBelowSG {
// only when user query with alias, the alias in path will be set
path.setMeasurementAlias(node.getAlias());
}
+ if (withTags) {
+ try {
+ path.setTagMap(tagManager.readTagFile(node.getOffset()).getLeft());
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
result.add(path);
}
};
diff --git a/server/src/main/java/org/apache/iotdb/db/metadata/mtree/MTreeBelowSGMemoryImpl.java b/server/src/main/java/org/apache/iotdb/db/metadata/mtree/MTreeBelowSGMemoryImpl.java
index 4151063116..f106906702 100644
--- a/server/src/main/java/org/apache/iotdb/db/metadata/mtree/MTreeBelowSGMemoryImpl.java
+++ b/server/src/main/java/org/apache/iotdb/db/metadata/mtree/MTreeBelowSGMemoryImpl.java
@@ -52,6 +52,7 @@ import org.apache.iotdb.db.metadata.mtree.traverser.counter.MNodeLevelCounter;
import org.apache.iotdb.db.metadata.mtree.traverser.counter.MeasurementCounter;
import org.apache.iotdb.db.metadata.mtree.traverser.counter.MeasurementGroupByLevelCounter;
import org.apache.iotdb.db.metadata.path.MeasurementPath;
+import org.apache.iotdb.db.metadata.tag.TagManager;
import org.apache.iotdb.db.metadata.template.Template;
import org.apache.iotdb.db.metadata.utils.MetaFormatUtils;
import org.apache.iotdb.db.qp.physical.sys.ShowDevicesPlan;
@@ -120,23 +121,30 @@ public class MTreeBelowSGMemoryImpl implements IMTreeBelowSG {
// this implementation is based on memory, thus only MTree write operation must invoke MTreeStore
private MemMTreeStore store;
private volatile IStorageGroupMNode storageGroupMNode;
+ private final TagManager tagManager;
private int levelOfSG;
// region MTree initialization, clear and serialization
- public MTreeBelowSGMemoryImpl(IStorageGroupMNode storageGroupMNode, int schemaRegionId) {
+ public MTreeBelowSGMemoryImpl(
+ IStorageGroupMNode storageGroupMNode, TagManager tagManager, int schemaRegionId) {
PartialPath storageGroup = storageGroupMNode.getPartialPath();
store = new MemMTreeStore(storageGroup, true);
this.storageGroupMNode = store.getRoot().getAsStorageGroupMNode();
this.storageGroupMNode.setParent(storageGroupMNode.getParent());
levelOfSG = storageGroup.getNodeLength() - 1;
+ this.tagManager = tagManager;
}
private MTreeBelowSGMemoryImpl(
- MemMTreeStore store, IStorageGroupMNode storageGroupMNode, int schemaRegionId) {
+ MemMTreeStore store,
+ IStorageGroupMNode storageGroupMNode,
+ TagManager tagManager,
+ int schemaRegionId) {
this.store = store;
this.storageGroupMNode = store.getRoot().getAsStorageGroupMNode();
this.storageGroupMNode.setParent(storageGroupMNode.getParent());
levelOfSG = storageGroupMNode.getPartialPath().getNodeLength() - 1;
+ this.tagManager = tagManager;
}
@Override
@@ -153,11 +161,13 @@ public class MTreeBelowSGMemoryImpl implements IMTreeBelowSG {
File snapshotDir,
IStorageGroupMNode storageGroupMNode,
int schemaRegionId,
- Consumer<IMeasurementMNode> measurementProcess)
+ Consumer<IMeasurementMNode> measurementProcess,
+ TagManager tagManager)
throws IOException {
return new MTreeBelowSGMemoryImpl(
MemMTreeStore.loadFromSnapshot(snapshotDir, measurementProcess),
storageGroupMNode,
+ tagManager,
schemaRegionId);
}
@@ -620,7 +630,7 @@ public class MTreeBelowSGMemoryImpl implements IMTreeBelowSG {
@Override
public List<MeasurementPath> getMeasurementPaths(PartialPath pathPattern, boolean isPrefixMatch)
throws MetadataException {
- return getMeasurementPathsWithAlias(pathPattern, 0, 0, isPrefixMatch).left;
+ return getMeasurementPathsWithAlias(pathPattern, 0, 0, isPrefixMatch, false).left;
}
/**
@@ -646,7 +656,7 @@ public class MTreeBelowSGMemoryImpl implements IMTreeBelowSG {
*/
@Override
public Pair<List<MeasurementPath>, Integer> getMeasurementPathsWithAlias(
- PartialPath pathPattern, int limit, int offset, boolean isPrefixMatch)
+ PartialPath pathPattern, int limit, int offset, boolean isPrefixMatch, boolean withTags)
throws MetadataException {
List<MeasurementPath> result = new LinkedList<>();
MeasurementCollector<List<PartialPath>> collector =
@@ -659,6 +669,13 @@ public class MTreeBelowSGMemoryImpl implements IMTreeBelowSG {
// only when user query with alias, the alias in path will be set
path.setMeasurementAlias(node.getAlias());
}
+ if (withTags) {
+ try {
+ path.setTagMap(tagManager.readTagFile(node.getOffset()).getLeft());
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
result.add(path);
}
};
@@ -669,7 +686,8 @@ public class MTreeBelowSGMemoryImpl implements IMTreeBelowSG {
}
public List<MeasurementPath> fetchSchema(
- PartialPath pathPattern, Map<Integer, Template> templateMap) throws MetadataException {
+ PartialPath pathPattern, Map<Integer, Template> templateMap, boolean withTags)
+ throws MetadataException {
List<MeasurementPath> result = new LinkedList<>();
MeasurementCollector<List<PartialPath>> collector =
new MeasurementCollector<List<PartialPath>>(storageGroupMNode, pathPattern, store) {
@@ -680,6 +698,13 @@ public class MTreeBelowSGMemoryImpl implements IMTreeBelowSG {
// only when user query with alias, the alias in path will be set
path.setMeasurementAlias(node.getAlias());
}
+ if (withTags) {
+ try {
+ path.setTagMap(tagManager.readTagFile(node.getOffset()).getLeft());
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
result.add(path);
}
};
@@ -1274,6 +1299,7 @@ public class MTreeBelowSGMemoryImpl implements IMTreeBelowSG {
* @return index on full path of the node which matches all measurements' path with its
* upperTemplate.
*/
+ @Override
public int getMountedNodeIndexOnMeasurementPath(PartialPath devicePath, String[] measurements)
throws MetadataException {
String[] nodes = devicePath.getNodes();
diff --git a/server/src/main/java/org/apache/iotdb/db/metadata/path/MeasurementPath.java b/server/src/main/java/org/apache/iotdb/db/metadata/path/MeasurementPath.java
index 2525844e31..3872d49a5c 100644
--- a/server/src/main/java/org/apache/iotdb/db/metadata/path/MeasurementPath.java
+++ b/server/src/main/java/org/apache/iotdb/db/metadata/path/MeasurementPath.java
@@ -34,6 +34,8 @@ import org.slf4j.LoggerFactory;
import java.io.DataOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
+import java.util.HashMap;
+import java.util.Map;
public class MeasurementPath extends PartialPath {
@@ -41,6 +43,8 @@ public class MeasurementPath extends PartialPath {
private IMeasurementSchema measurementSchema;
+ private Map<String, String> tagMap;
+
private boolean isUnderAlignedEntity = false;
// alias of measurement, null pointer cannot be serialized in thrift so empty string is instead
@@ -86,6 +90,10 @@ public class MeasurementPath extends PartialPath {
return measurementSchema;
}
+ public Map<String, String> getTagMap() {
+ return tagMap;
+ }
+
@Override
public TSDataType getSeriesType() {
return getMeasurementSchema().getType();
@@ -99,6 +107,10 @@ public class MeasurementPath extends PartialPath {
this.measurementSchema = measurementSchema;
}
+ public void setTagMap(Map<String, String> tagMap) {
+ this.tagMap = tagMap;
+ }
+
@Override
public String getMeasurementAlias() {
return measurementAlias;
@@ -141,6 +153,9 @@ public class MeasurementPath extends PartialPath {
result.device = device;
result.measurementAlias = measurementAlias;
result.measurementSchema = measurementSchema;
+ if (tagMap != null) {
+ result.tagMap = new HashMap<>(tagMap);
+ }
result.isUnderAlignedEntity = isUnderAlignedEntity;
return result;
}
@@ -160,6 +175,10 @@ public class MeasurementPath extends PartialPath {
newMeasurementPath =
new MeasurementPath(this.getDevice(), this.getMeasurement(), this.getMeasurementSchema());
newMeasurementPath.setUnderAlignedEntity(this.isUnderAlignedEntity);
+ newMeasurementPath.setMeasurementAlias(this.measurementAlias);
+ if (tagMap != null) {
+ newMeasurementPath.setTagMap(new HashMap<>(tagMap));
+ }
} catch (IllegalPathException e) {
logger.warn("path is illegal: {}", this.getFullPath(), e);
}
@@ -181,6 +200,12 @@ public class MeasurementPath extends PartialPath {
}
measurementSchema.serializeTo(byteBuffer);
}
+ if (tagMap == null) {
+ ReadWriteIOUtils.write((byte) 0, byteBuffer);
+ } else {
+ ReadWriteIOUtils.write((byte) 1, byteBuffer);
+ ReadWriteIOUtils.write(tagMap, byteBuffer);
+ }
ReadWriteIOUtils.write(isUnderAlignedEntity, byteBuffer);
ReadWriteIOUtils.write(measurementAlias, byteBuffer);
}
@@ -200,6 +225,12 @@ public class MeasurementPath extends PartialPath {
}
measurementSchema.serializeTo(stream);
}
+ if (tagMap == null) {
+ ReadWriteIOUtils.write((byte) 0, stream);
+ } else {
+ ReadWriteIOUtils.write((byte) 1, stream);
+ ReadWriteIOUtils.write(tagMap, stream);
+ }
ReadWriteIOUtils.write(isUnderAlignedEntity, stream);
ReadWriteIOUtils.write(measurementAlias, stream);
}
@@ -216,6 +247,10 @@ public class MeasurementPath extends PartialPath {
measurementPath.measurementSchema = VectorMeasurementSchema.deserializeFrom(byteBuffer);
}
}
+ isNull = ReadWriteIOUtils.readByte(byteBuffer);
+ if (isNull == 1) {
+ measurementPath.tagMap = ReadWriteIOUtils.readMap(byteBuffer);
+ }
measurementPath.isUnderAlignedEntity = ReadWriteIOUtils.readBool(byteBuffer);
measurementPath.measurementAlias = ReadWriteIOUtils.readString(byteBuffer);
measurementPath.nodes = partialPath.getNodes();
diff --git a/server/src/main/java/org/apache/iotdb/db/metadata/schemaregion/ISchemaRegion.java b/server/src/main/java/org/apache/iotdb/db/metadata/schemaregion/ISchemaRegion.java
index 860d5b2005..ce87ca74cc 100644
--- a/server/src/main/java/org/apache/iotdb/db/metadata/schemaregion/ISchemaRegion.java
+++ b/server/src/main/java/org/apache/iotdb/db/metadata/schemaregion/ISchemaRegion.java
@@ -258,9 +258,10 @@ public interface ISchemaRegion {
*
* @param pathPattern can be a pattern or a full path of timeseries.
* @param isPrefixMatch if true, the path pattern is used to match prefix path
+ * @param withTags whether returns tag kvs in the result list.
*/
- List<MeasurementPath> getMeasurementPaths(PartialPath pathPattern, boolean isPrefixMatch)
- throws MetadataException;
+ List<MeasurementPath> getMeasurementPaths(
+ PartialPath pathPattern, boolean isPrefixMatch, boolean withTags) throws MetadataException;
/**
* Similar to method getMeasurementPaths(), but return Path with alias and filter the result by
@@ -270,10 +271,11 @@ public interface ISchemaRegion {
* @param isPrefixMatch if true, the path pattern is used to match prefix path
*/
Pair<List<MeasurementPath>, Integer> getMeasurementPathsWithAlias(
- PartialPath pathPattern, int limit, int offset, boolean isPrefixMatch)
+ PartialPath pathPattern, int limit, int offset, boolean isPrefixMatch, boolean withTags)
throws MetadataException;
- List<MeasurementPath> fetchSchema(PartialPath pathPattern, Map<Integer, Template> templateMap)
+ List<MeasurementPath> fetchSchema(
+ PartialPath pathPattern, Map<Integer, Template> templateMap, boolean withTags)
throws MetadataException;
Pair<List<ShowTimeSeriesResult>, Integer> showTimeseries(
diff --git a/server/src/main/java/org/apache/iotdb/db/metadata/schemaregion/SchemaRegionMemoryImpl.java b/server/src/main/java/org/apache/iotdb/db/metadata/schemaregion/SchemaRegionMemoryImpl.java
index 90394b6041..274c975f44 100644
--- a/server/src/main/java/org/apache/iotdb/db/metadata/schemaregion/SchemaRegionMemoryImpl.java
+++ b/server/src/main/java/org/apache/iotdb/db/metadata/schemaregion/SchemaRegionMemoryImpl.java
@@ -224,6 +224,7 @@ public class SchemaRegionMemoryImpl implements ISchemaRegion {
init();
}
+ @Override
@SuppressWarnings("squid:S2093")
public synchronized void init() throws MetadataException {
if (initialized) {
@@ -239,7 +240,7 @@ public class SchemaRegionMemoryImpl implements ISchemaRegion {
isRecovering = true;
tagManager = new TagManager(schemaRegionDirPath);
- mtree = new MTreeBelowSGMemoryImpl(storageGroupMNode, schemaRegionId.getId());
+ mtree = new MTreeBelowSGMemoryImpl(storageGroupMNode, tagManager, schemaRegionId.getId());
if (!(config.isClusterMode()
&& config
@@ -300,6 +301,7 @@ public class SchemaRegionMemoryImpl implements ISchemaRegion {
}
}
+ @Override
public void forceMlog() {
if (!initialized) {
return;
@@ -449,14 +451,17 @@ public class SchemaRegionMemoryImpl implements ISchemaRegion {
// region Interfaces for schema region Info query and operation
+ @Override
public String getStorageGroupFullPath() {
return storageGroupFullPath;
}
+ @Override
public SchemaRegionId getSchemaRegionId() {
return schemaRegionId;
}
+ @Override
public synchronized void deleteSchemaRegion() throws MetadataException {
// collect all the LeafMNode in this schema region
List<IMeasurementMNode> leafMNodes = mtree.getAllMeasurementMNode();
@@ -543,7 +548,8 @@ public class SchemaRegionMemoryImpl implements ISchemaRegion {
storageGroupFullPath + PATH_SEPARATOR + measurementMNode.getFullPath(),
schemaRegionId);
}
- });
+ },
+ tagManager);
logger.info(
"MTree snapshot loading of schemaRegion {} costs {}ms.",
schemaRegionId,
@@ -585,6 +591,7 @@ public class SchemaRegionMemoryImpl implements ISchemaRegion {
createTimeseries(plan, -1);
}
+ @Override
@SuppressWarnings("squid:S3776") // Suppress high Cognitive Complexity warning
public void createTimeseries(CreateTimeSeriesPlan plan, long offset) throws MetadataException {
if (!memoryStatistics.isAllowToCreateNewSeries()) {
@@ -705,6 +712,7 @@ public class SchemaRegionMemoryImpl implements ISchemaRegion {
*
* @param plan CreateAlignedTimeSeriesPlan
*/
+ @Override
public void createAlignedTimeSeries(CreateAlignedTimeSeriesPlan plan) throws MetadataException {
int seriesCount = plan.getMeasurements().size();
if (!memoryStatistics.isAllowToCreateNewSeries()) {
@@ -814,6 +822,7 @@ public class SchemaRegionMemoryImpl implements ISchemaRegion {
* @param isPrefixMatch if true, the path pattern is used to match prefix path
* @return deletion failed Timeseries
*/
+ @Override
public synchronized Pair<Integer, Set<String>> deleteTimeseries(
PartialPath pathPattern, boolean isPrefixMatch) throws MetadataException {
try {
@@ -914,6 +923,7 @@ public class SchemaRegionMemoryImpl implements ISchemaRegion {
return node;
}
+ @Override
public void autoCreateDeviceMNode(AutoCreateDeviceMNodePlan plan) throws MetadataException {
mtree.getDeviceNodeWithAutoCreating(plan.getPath());
try {
@@ -930,6 +940,7 @@ public class SchemaRegionMemoryImpl implements ISchemaRegion {
*
* @param path a full path or a prefix path
*/
+ @Override
public boolean isPathExist(PartialPath path) {
try {
return mtree.isPathExist(path);
@@ -946,6 +957,7 @@ public class SchemaRegionMemoryImpl implements ISchemaRegion {
* path, may contain wildcard. If using prefix match, the path pattern is used to match prefix
* path. All timeseries start with the matched prefix path will be counted.
*/
+ @Override
public int getAllTimeseriesCount(PartialPath pathPattern, boolean isPrefixMatch)
throws MetadataException {
return mtree.getAllTimeseriesCount(pathPattern, isPrefixMatch);
@@ -974,6 +986,7 @@ public class SchemaRegionMemoryImpl implements ISchemaRegion {
* pattern is used to match prefix path. All timeseries start with the matched prefix path will be
* counted.
*/
+ @Override
public int getDevicesNum(PartialPath pathPattern, boolean isPrefixMatch)
throws MetadataException {
return mtree.getDevicesNum(pathPattern, isPrefixMatch);
@@ -993,11 +1006,13 @@ public class SchemaRegionMemoryImpl implements ISchemaRegion {
* @param level the level should match the level of the path
* @param isPrefixMatch if true, the path pattern is used to match prefix path
*/
+ @Override
public int getNodesCountInGivenLevel(PartialPath pathPattern, int level, boolean isPrefixMatch)
throws MetadataException {
return mtree.getNodesCountInGivenLevel(pathPattern, level, isPrefixMatch);
}
+ @Override
public Map<PartialPath, Integer> getMeasurementCountGroupByLevel(
PartialPath pathPattern, int level, boolean isPrefixMatch) throws MetadataException {
return mtree.getMeasurementCountGroupByLevel(pathPattern, level, isPrefixMatch);
@@ -1023,6 +1038,7 @@ public class SchemaRegionMemoryImpl implements ISchemaRegion {
// endregion
// region Interfaces for level Node info Query
+ @Override
public List<PartialPath> getNodesListInGivenLevel(
PartialPath pathPattern,
int nodeLevel,
@@ -1043,6 +1059,7 @@ public class SchemaRegionMemoryImpl implements ISchemaRegion {
* @param pathPattern The given path
* @return All child nodes' seriesPath(s) of given seriesPath.
*/
+ @Override
public Set<TSchemaNode> getChildNodePathInNextLevel(PartialPath pathPattern)
throws MetadataException {
return mtree.getChildNodePathInNextLevel(pathPattern);
@@ -1058,6 +1075,7 @@ public class SchemaRegionMemoryImpl implements ISchemaRegion {
*
* @return All child nodes of given seriesPath.
*/
+ @Override
public Set<String> getChildNodeNameInNextLevel(PartialPath pathPattern) throws MetadataException {
return mtree.getChildNodeNameInNextLevel(pathPattern);
}
@@ -1072,6 +1090,7 @@ public class SchemaRegionMemoryImpl implements ISchemaRegion {
* @param timeseries a path pattern of the target timeseries
* @return A HashSet instance which stores devices paths.
*/
+ @Override
public Set<PartialPath> getBelongedDevices(PartialPath timeseries) throws MetadataException {
return mtree.getDevicesByTimeseries(timeseries);
}
@@ -1084,6 +1103,7 @@ public class SchemaRegionMemoryImpl implements ISchemaRegion {
* @param isPrefixMatch if true, the path pattern is used to match prefix path.
* @return A HashSet instance which stores devices paths matching the given path pattern.
*/
+ @Override
public Set<PartialPath> getMatchedDevices(PartialPath pathPattern, boolean isPrefixMatch)
throws MetadataException {
return mtree.getDevices(pathPattern, isPrefixMatch);
@@ -1095,6 +1115,7 @@ public class SchemaRegionMemoryImpl implements ISchemaRegion {
* @param plan ShowDevicesPlan which contains the path pattern and restriction params.
* @return ShowDevicesResult and the current offset of this region after traverse.
*/
+ @Override
public Pair<List<ShowDevicesResult>, Integer> getMatchedDevices(ShowDevicesPlan plan)
throws MetadataException {
return mtree.getDevices(plan);
@@ -1112,9 +1133,10 @@ public class SchemaRegionMemoryImpl implements ISchemaRegion {
* @param pathPattern can be a pattern or a full path of timeseries.
* @param isPrefixMatch if true, the path pattern is used to match prefix path
*/
- public List<MeasurementPath> getMeasurementPaths(PartialPath pathPattern, boolean isPrefixMatch)
- throws MetadataException {
- return getMeasurementPathsWithAlias(pathPattern, 0, 0, isPrefixMatch).left;
+ @Override
+ public List<MeasurementPath> getMeasurementPaths(
+ PartialPath pathPattern, boolean isPrefixMatch, boolean withTags) throws MetadataException {
+ return getMeasurementPathsWithAlias(pathPattern, 0, 0, isPrefixMatch, withTags).left;
}
/**
@@ -1126,7 +1148,7 @@ public class SchemaRegionMemoryImpl implements ISchemaRegion {
*/
public List<MeasurementPath> getMeasurementPaths(PartialPath pathPattern)
throws MetadataException {
- return getMeasurementPaths(pathPattern, false);
+ return getMeasurementPaths(pathPattern, false, false);
}
/**
@@ -1136,18 +1158,21 @@ public class SchemaRegionMemoryImpl implements ISchemaRegion {
*
* @param isPrefixMatch if true, the path pattern is used to match prefix path
*/
+ @Override
public Pair<List<MeasurementPath>, Integer> getMeasurementPathsWithAlias(
- PartialPath pathPattern, int limit, int offset, boolean isPrefixMatch)
+ PartialPath pathPattern, int limit, int offset, boolean isPrefixMatch, boolean withTags)
throws MetadataException {
- return mtree.getMeasurementPathsWithAlias(pathPattern, limit, offset, isPrefixMatch);
+ return mtree.getMeasurementPathsWithAlias(pathPattern, limit, offset, isPrefixMatch, withTags);
}
@Override
public List<MeasurementPath> fetchSchema(
- PartialPath pathPattern, Map<Integer, Template> templateMap) throws MetadataException {
- return mtree.fetchSchema(pathPattern, templateMap);
+ PartialPath pathPattern, Map<Integer, Template> templateMap, boolean withTags)
+ throws MetadataException {
+ return mtree.fetchSchema(pathPattern, templateMap, withTags);
}
+ @Override
public Pair<List<ShowTimeSeriesResult>, Integer> showTimeseries(
ShowTimeSeriesPlan plan, QueryContext context) throws MetadataException {
// show timeseries with index
@@ -1292,6 +1317,7 @@ public class SchemaRegionMemoryImpl implements ISchemaRegion {
}
// attention: this path must be a device node
+ @Override
public List<MeasurementPath> getAllMeasurementByDevicePath(PartialPath devicePath)
throws PathNotExistException {
List<MeasurementPath> res = new LinkedList<>();
@@ -1329,6 +1355,7 @@ public class SchemaRegionMemoryImpl implements ISchemaRegion {
// endregion
// region Interfaces and methods for MNode query
+ @Override
public IMNode getDeviceNode(PartialPath path) throws MetadataException {
IMNode node;
try {
@@ -1342,6 +1369,7 @@ public class SchemaRegionMemoryImpl implements ISchemaRegion {
}
}
+ @Override
public IMeasurementMNode getMeasurementMNode(PartialPath fullPath) throws MetadataException {
return mtree.getMeasurementMNode(fullPath);
}
@@ -1388,6 +1416,7 @@ public class SchemaRegionMemoryImpl implements ISchemaRegion {
}
}
+ @Override
public void changeAlias(PartialPath path, String alias) throws MetadataException {
IMeasurementMNode leafMNode = mtree.getMeasurementMNode(path);
if (leafMNode.getAlias() != null) {
@@ -1412,6 +1441,7 @@ public class SchemaRegionMemoryImpl implements ISchemaRegion {
* @param attributesMap newly added attributes map
* @param fullPath timeseries
*/
+ @Override
@SuppressWarnings("squid:S3776") // Suppress high Cognitive Complexity warning
public void upsertTagsAndAttributes(
String alias,
@@ -1467,6 +1497,7 @@ public class SchemaRegionMemoryImpl implements ISchemaRegion {
* @param attributesMap newly added attributes map
* @param fullPath timeseries
*/
+ @Override
public void addAttributes(Map<String, String> attributesMap, PartialPath fullPath)
throws MetadataException, IOException {
IMeasurementMNode leafMNode = mtree.getMeasurementMNode(fullPath);
@@ -1488,6 +1519,7 @@ public class SchemaRegionMemoryImpl implements ISchemaRegion {
* @param tagsMap newly added tags map
* @param fullPath timeseries
*/
+ @Override
public void addTags(Map<String, String> tagsMap, PartialPath fullPath)
throws MetadataException, IOException {
IMeasurementMNode leafMNode = mtree.getMeasurementMNode(fullPath);
@@ -1510,6 +1542,7 @@ public class SchemaRegionMemoryImpl implements ISchemaRegion {
* @param keySet tags key or attributes key
* @param fullPath timeseries path
*/
+ @Override
@SuppressWarnings("squid:S3776") // Suppress high Cognitive Complexity warning
public void dropTagsOrAttributes(Set<String> keySet, PartialPath fullPath)
throws MetadataException, IOException {
@@ -1526,6 +1559,7 @@ public class SchemaRegionMemoryImpl implements ISchemaRegion {
* @param alterMap the new tags or attributes key-value
* @param fullPath timeseries
*/
+ @Override
@SuppressWarnings("squid:S3776") // Suppress high Cognitive Complexity warning
public void setTagsOrAttributesValue(Map<String, String> alterMap, PartialPath fullPath)
throws MetadataException, IOException {
@@ -1546,6 +1580,7 @@ public class SchemaRegionMemoryImpl implements ISchemaRegion {
* @param newKey new key of tag or attribute
* @param fullPath timeseries
*/
+ @Override
@SuppressWarnings("squid:S3776") // Suppress high Cognitive Complexity warning
public void renameTagOrAttributeKey(String oldKey, String newKey, PartialPath fullPath)
throws MetadataException, IOException {
@@ -1568,6 +1603,7 @@ public class SchemaRegionMemoryImpl implements ISchemaRegion {
// region Interfaces and Implementation for InsertPlan process
/** get schema for device. Attention!!! Only support insertPlan */
+ @Override
@SuppressWarnings("squid:S3776") // Suppress high Cognitive Complexity warning
public IMNode getSeriesSchemasAndReadLockDevice(InsertPlan plan)
throws MetadataException, IOException {
@@ -1834,19 +1870,23 @@ public class SchemaRegionMemoryImpl implements ISchemaRegion {
* @param templateName designated template name, blank string for any template exists
* @return paths set
*/
+ @Override
public Set<String> getPathsSetTemplate(String templateName) throws MetadataException {
return new HashSet<>(mtree.getPathsSetOnTemplate(templateName));
}
+ @Override
public Set<String> getPathsUsingTemplate(String templateName) throws MetadataException {
return new HashSet<>(mtree.getPathsUsingTemplate(templateName));
}
+ @Override
public boolean isTemplateAppendable(Template template, List<String> measurements)
throws MetadataException {
return mtree.isTemplateAppendable(template, measurements);
}
+ @Override
public synchronized void setSchemaTemplate(SetTemplatePlan plan) throws MetadataException {
// get mnode and update template should be atomic
Template template = TemplateManager.getInstance().getTemplate(plan.getTemplateName());
@@ -1872,6 +1912,7 @@ public class SchemaRegionMemoryImpl implements ISchemaRegion {
}
}
+ @Override
public synchronized void unsetSchemaTemplate(UnsetTemplatePlan plan) throws MetadataException {
// get mnode should be atomic
try {
@@ -1896,6 +1937,7 @@ public class SchemaRegionMemoryImpl implements ISchemaRegion {
}
}
+ @Override
public void setUsingSchemaTemplate(ActivateTemplatePlan plan) throws MetadataException {
// check whether any template has been set on designated path
if (mtree.getTemplateOnPath(plan.getPrefixPath()) == null) {
@@ -1993,10 +2035,12 @@ public class SchemaRegionMemoryImpl implements ISchemaRegion {
// region Interfaces for Trigger
+ @Override
public IMNode getMNodeForTrigger(PartialPath fullPath) throws MetadataException {
return mtree.getNodeByPath(fullPath);
}
+ @Override
public void releaseMNodeAfterDropTrigger(IMNode node) throws MetadataException {
// do nothing
}
diff --git a/server/src/main/java/org/apache/iotdb/db/metadata/schemaregion/SchemaRegionSchemaFileImpl.java b/server/src/main/java/org/apache/iotdb/db/metadata/schemaregion/SchemaRegionSchemaFileImpl.java
index 051f78898f..7ca40c8b6f 100644
--- a/server/src/main/java/org/apache/iotdb/db/metadata/schemaregion/SchemaRegionSchemaFileImpl.java
+++ b/server/src/main/java/org/apache/iotdb/db/metadata/schemaregion/SchemaRegionSchemaFileImpl.java
@@ -209,6 +209,7 @@ public class SchemaRegionSchemaFileImpl implements ISchemaRegion {
init();
}
+ @Override
@SuppressWarnings("squid:S2093")
public synchronized void init() throws MetadataException {
if (initialized) {
@@ -254,7 +255,7 @@ public class SchemaRegionSchemaFileImpl implements ISchemaRegion {
isRecovering = true;
tagManager = new TagManager(schemaRegionDirPath);
- mtree = new MTreeBelowSGCachedImpl(storageGroupMNode, schemaRegionId.getId());
+ mtree = new MTreeBelowSGCachedImpl(storageGroupMNode, tagManager, schemaRegionId.getId());
int lineNumber = initFromLog(logFile);
@@ -270,6 +271,7 @@ public class SchemaRegionSchemaFileImpl implements ISchemaRegion {
initialized = true;
}
+ @Override
public void forceMlog() {
if (!initialized) {
return;
@@ -407,14 +409,17 @@ public class SchemaRegionSchemaFileImpl implements ISchemaRegion {
// region Interfaces for schema region Info query and operation
+ @Override
public String getStorageGroupFullPath() {
return storageGroupFullPath;
}
+ @Override
public SchemaRegionId getSchemaRegionId() {
return schemaRegionId;
}
+ @Override
public synchronized void deleteSchemaRegion() throws MetadataException {
// collect all the LeafMNode in this schema region
List<IMeasurementMNode> leafMNodes = mtree.getAllMeasurementMNode();
@@ -475,6 +480,7 @@ public class SchemaRegionSchemaFileImpl implements ISchemaRegion {
}
}
+ @Override
@SuppressWarnings("squid:S3776") // Suppress high Cognitive Complexity warning
public void createTimeseries(CreateTimeSeriesPlan plan, long offset) throws MetadataException {
if (!memoryStatistics.isAllowToCreateNewSeries()) {
@@ -621,6 +627,7 @@ public class SchemaRegionSchemaFileImpl implements ISchemaRegion {
*
* @param plan CreateAlignedTimeSeriesPlan
*/
+ @Override
public void createAlignedTimeSeries(CreateAlignedTimeSeriesPlan plan) throws MetadataException {
int seriesCount = plan.getMeasurements().size();
if (!memoryStatistics.isAllowToCreateNewSeries()) {
@@ -736,6 +743,7 @@ public class SchemaRegionSchemaFileImpl implements ISchemaRegion {
* @param isPrefixMatch if true, the path pattern is used to match prefix path
* @return deletion failed Timeseries
*/
+ @Override
public synchronized Pair<Integer, Set<String>> deleteTimeseries(
PartialPath pathPattern, boolean isPrefixMatch) throws MetadataException {
try {
@@ -849,6 +857,7 @@ public class SchemaRegionSchemaFileImpl implements ISchemaRegion {
return node;
}
+ @Override
public void autoCreateDeviceMNode(AutoCreateDeviceMNodePlan plan) throws MetadataException {
IMNode node = mtree.getDeviceNodeWithAutoCreating(plan.getPath());
mtree.unPinMNode(node);
@@ -868,6 +877,7 @@ public class SchemaRegionSchemaFileImpl implements ISchemaRegion {
*
* @param path a full path or a prefix path
*/
+ @Override
public boolean isPathExist(PartialPath path) {
try {
return mtree.isPathExist(path);
@@ -884,6 +894,7 @@ public class SchemaRegionSchemaFileImpl implements ISchemaRegion {
* path, may contain wildcard. If using prefix match, the path pattern is used to match prefix
* path. All timeseries start with the matched prefix path will be counted.
*/
+ @Override
public int getAllTimeseriesCount(PartialPath pathPattern, boolean isPrefixMatch)
throws MetadataException {
return mtree.getAllTimeseriesCount(pathPattern, isPrefixMatch);
@@ -912,6 +923,7 @@ public class SchemaRegionSchemaFileImpl implements ISchemaRegion {
* pattern is used to match prefix path. All timeseries start with the matched prefix path will be
* counted.
*/
+ @Override
public int getDevicesNum(PartialPath pathPattern, boolean isPrefixMatch)
throws MetadataException {
return mtree.getDevicesNum(pathPattern, isPrefixMatch);
@@ -931,11 +943,13 @@ public class SchemaRegionSchemaFileImpl implements ISchemaRegion {
* @param level the level should match the level of the path
* @param isPrefixMatch if true, the path pattern is used to match prefix path
*/
+ @Override
public int getNodesCountInGivenLevel(PartialPath pathPattern, int level, boolean isPrefixMatch)
throws MetadataException {
return mtree.getNodesCountInGivenLevel(pathPattern, level, isPrefixMatch);
}
+ @Override
public Map<PartialPath, Integer> getMeasurementCountGroupByLevel(
PartialPath pathPattern, int level, boolean isPrefixMatch) throws MetadataException {
return mtree.getMeasurementCountGroupByLevel(pathPattern, level, isPrefixMatch);
@@ -961,6 +975,7 @@ public class SchemaRegionSchemaFileImpl implements ISchemaRegion {
// endregion
// region Interfaces for level Node info Query
+ @Override
public List<PartialPath> getNodesListInGivenLevel(
PartialPath pathPattern,
int nodeLevel,
@@ -981,6 +996,7 @@ public class SchemaRegionSchemaFileImpl implements ISchemaRegion {
* @param pathPattern The given path
* @return All child nodes' seriesPath(s) of given seriesPath.
*/
+ @Override
public Set<TSchemaNode> getChildNodePathInNextLevel(PartialPath pathPattern)
throws MetadataException {
return mtree.getChildNodePathInNextLevel(pathPattern);
@@ -996,6 +1012,7 @@ public class SchemaRegionSchemaFileImpl implements ISchemaRegion {
*
* @return All child nodes of given seriesPath.
*/
+ @Override
public Set<String> getChildNodeNameInNextLevel(PartialPath pathPattern) throws MetadataException {
return mtree.getChildNodeNameInNextLevel(pathPattern);
}
@@ -1010,6 +1027,7 @@ public class SchemaRegionSchemaFileImpl implements ISchemaRegion {
* @param timeseries a path pattern of the target timeseries
* @return A HashSet instance which stores devices paths.
*/
+ @Override
public Set<PartialPath> getBelongedDevices(PartialPath timeseries) throws MetadataException {
return mtree.getDevicesByTimeseries(timeseries);
}
@@ -1022,6 +1040,7 @@ public class SchemaRegionSchemaFileImpl implements ISchemaRegion {
* @param isPrefixMatch if true, the path pattern is used to match prefix path.
* @return A HashSet instance which stores devices paths matching the given path pattern.
*/
+ @Override
public Set<PartialPath> getMatchedDevices(PartialPath pathPattern, boolean isPrefixMatch)
throws MetadataException {
return mtree.getDevices(pathPattern, isPrefixMatch);
@@ -1033,6 +1052,7 @@ public class SchemaRegionSchemaFileImpl implements ISchemaRegion {
* @param plan ShowDevicesPlan which contains the path pattern and restriction params.
* @return ShowDevicesResult and the current offset of this region after traverse.
*/
+ @Override
public Pair<List<ShowDevicesResult>, Integer> getMatchedDevices(ShowDevicesPlan plan)
throws MetadataException {
return mtree.getDevices(plan);
@@ -1050,9 +1070,10 @@ public class SchemaRegionSchemaFileImpl implements ISchemaRegion {
* @param pathPattern can be a pattern or a full path of timeseries.
* @param isPrefixMatch if true, the path pattern is used to match prefix path
*/
- public List<MeasurementPath> getMeasurementPaths(PartialPath pathPattern, boolean isPrefixMatch)
- throws MetadataException {
- return getMeasurementPathsWithAlias(pathPattern, 0, 0, isPrefixMatch).left;
+ @Override
+ public List<MeasurementPath> getMeasurementPaths(
+ PartialPath pathPattern, boolean isPrefixMatch, boolean withTags) throws MetadataException {
+ return getMeasurementPathsWithAlias(pathPattern, 0, 0, isPrefixMatch, withTags).left;
}
/**
@@ -1062,18 +1083,21 @@ public class SchemaRegionSchemaFileImpl implements ISchemaRegion {
*
* @param isPrefixMatch if true, the path pattern is used to match prefix path
*/
+ @Override
public Pair<List<MeasurementPath>, Integer> getMeasurementPathsWithAlias(
- PartialPath pathPattern, int limit, int offset, boolean isPrefixMatch)
+ PartialPath pathPattern, int limit, int offset, boolean isPrefixMatch, boolean withTags)
throws MetadataException {
- return mtree.getMeasurementPathsWithAlias(pathPattern, limit, offset, isPrefixMatch);
+ return mtree.getMeasurementPathsWithAlias(pathPattern, limit, offset, isPrefixMatch, withTags);
}
@Override
public List<MeasurementPath> fetchSchema(
- PartialPath pathPattern, Map<Integer, Template> templateMap) throws MetadataException {
+ PartialPath pathPattern, Map<Integer, Template> templateMap, boolean withTags)
+ throws MetadataException {
throw new UnsupportedOperationException();
}
+ @Override
public Pair<List<ShowTimeSeriesResult>, Integer> showTimeseries(
ShowTimeSeriesPlan plan, QueryContext context) throws MetadataException {
// show timeseries with index
@@ -1218,6 +1242,7 @@ public class SchemaRegionSchemaFileImpl implements ISchemaRegion {
}
// attention: this path must be a device node
+ @Override
public List<MeasurementPath> getAllMeasurementByDevicePath(PartialPath devicePath)
throws PathNotExistException {
List<MeasurementPath> res = new LinkedList<>();
@@ -1256,6 +1281,7 @@ public class SchemaRegionSchemaFileImpl implements ISchemaRegion {
// region Interfaces and methods for MNode query
+ @Override
public IMNode getDeviceNode(PartialPath path) throws MetadataException {
IMNode node;
try {
@@ -1269,6 +1295,7 @@ public class SchemaRegionSchemaFileImpl implements ISchemaRegion {
}
}
+ @Override
public IMeasurementMNode getMeasurementMNode(PartialPath fullPath) throws MetadataException {
IMeasurementMNode measurementMNode = mtree.getMeasurementMNode(fullPath);
mtree.unPinMNode(measurementMNode);
@@ -1326,6 +1353,7 @@ public class SchemaRegionSchemaFileImpl implements ISchemaRegion {
}
}
+ @Override
public void changeAlias(PartialPath path, String alias) throws MetadataException {
IMeasurementMNode leafMNode = mtree.getMeasurementMNode(path);
try {
@@ -1356,6 +1384,7 @@ public class SchemaRegionSchemaFileImpl implements ISchemaRegion {
* @param attributesMap newly added attributes map
* @param fullPath timeseries
*/
+ @Override
@SuppressWarnings("squid:S3776") // Suppress high Cognitive Complexity warning
public void upsertTagsAndAttributes(
String alias,
@@ -1416,6 +1445,7 @@ public class SchemaRegionSchemaFileImpl implements ISchemaRegion {
* @param attributesMap newly added attributes map
* @param fullPath timeseries
*/
+ @Override
public void addAttributes(Map<String, String> attributesMap, PartialPath fullPath)
throws MetadataException, IOException {
IMeasurementMNode leafMNode = mtree.getMeasurementMNode(fullPath);
@@ -1441,6 +1471,7 @@ public class SchemaRegionSchemaFileImpl implements ISchemaRegion {
* @param tagsMap newly added tags map
* @param fullPath timeseries
*/
+ @Override
public void addTags(Map<String, String> tagsMap, PartialPath fullPath)
throws MetadataException, IOException {
IMeasurementMNode leafMNode = mtree.getMeasurementMNode(fullPath);
@@ -1469,6 +1500,7 @@ public class SchemaRegionSchemaFileImpl implements ISchemaRegion {
* @param keySet tags key or attributes key
* @param fullPath timeseries path
*/
+ @Override
@SuppressWarnings("squid:S3776") // Suppress high Cognitive Complexity warning
public void dropTagsOrAttributes(Set<String> keySet, PartialPath fullPath)
throws MetadataException, IOException {
@@ -1491,6 +1523,7 @@ public class SchemaRegionSchemaFileImpl implements ISchemaRegion {
* @param alterMap the new tags or attributes key-value
* @param fullPath timeseries
*/
+ @Override
@SuppressWarnings("squid:S3776") // Suppress high Cognitive Complexity warning
public void setTagsOrAttributesValue(Map<String, String> alterMap, PartialPath fullPath)
throws MetadataException, IOException {
@@ -1515,6 +1548,7 @@ public class SchemaRegionSchemaFileImpl implements ISchemaRegion {
* @param newKey new key of tag or attribute
* @param fullPath timeseries
*/
+ @Override
@SuppressWarnings("squid:S3776") // Suppress high Cognitive Complexity warning
public void renameTagOrAttributeKey(String oldKey, String newKey, PartialPath fullPath)
throws MetadataException, IOException {
@@ -1541,6 +1575,7 @@ public class SchemaRegionSchemaFileImpl implements ISchemaRegion {
// region Interfaces and Implementation for InsertPlan process
/** get schema for device. Attention!!! Only support insertPlan */
+ @Override
@SuppressWarnings("squid:S3776") // Suppress high Cognitive Complexity warning
public IMNode getSeriesSchemasAndReadLockDevice(InsertPlan plan)
throws MetadataException, IOException {
@@ -1781,19 +1816,23 @@ public class SchemaRegionSchemaFileImpl implements ISchemaRegion {
* @param templateName designated template name, blank string for any template exists
* @return paths set
*/
+ @Override
public Set<String> getPathsSetTemplate(String templateName) throws MetadataException {
return new HashSet<>(mtree.getPathsSetOnTemplate(templateName));
}
+ @Override
public Set<String> getPathsUsingTemplate(String templateName) throws MetadataException {
return new HashSet<>(mtree.getPathsUsingTemplate(templateName));
}
+ @Override
public boolean isTemplateAppendable(Template template, List<String> measurements)
throws MetadataException {
return mtree.isTemplateAppendable(template, measurements);
}
+ @Override
public synchronized void setSchemaTemplate(SetTemplatePlan plan) throws MetadataException {
// get mnode and update template should be atomic
Template template = TemplateManager.getInstance().getTemplate(plan.getTemplateName());
@@ -1826,6 +1865,7 @@ public class SchemaRegionSchemaFileImpl implements ISchemaRegion {
}
}
+ @Override
public synchronized void unsetSchemaTemplate(UnsetTemplatePlan plan) throws MetadataException {
// get mnode should be atomic
try {
@@ -1852,6 +1892,7 @@ public class SchemaRegionSchemaFileImpl implements ISchemaRegion {
}
}
+ @Override
public void setUsingSchemaTemplate(ActivateTemplatePlan plan) throws MetadataException {
// check whether any template has been set on designated path
if (mtree.getTemplateOnPath(plan.getPrefixPath()) == null) {
@@ -1930,10 +1971,12 @@ public class SchemaRegionSchemaFileImpl implements ISchemaRegion {
// region Interfaces for Trigger
+ @Override
public IMNode getMNodeForTrigger(PartialPath fullPath) throws MetadataException {
return mtree.getNodeByPath(fullPath);
}
+ @Override
public void releaseMNodeAfterDropTrigger(IMNode node) throws MetadataException {
mtree.unPinMNode(node);
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/common/schematree/ClusterSchemaTree.java b/server/src/main/java/org/apache/iotdb/db/mpp/common/schematree/ClusterSchemaTree.java
index ae5e7b97b5..c5d7804868 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/common/schematree/ClusterSchemaTree.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/common/schematree/ClusterSchemaTree.java
@@ -41,6 +41,7 @@ import java.util.ArrayDeque;
import java.util.ArrayList;
import java.util.Deque;
import java.util.List;
+import java.util.Map;
import static org.apache.iotdb.commons.conf.IoTDBConstant.PATH_ROOT;
import static org.apache.iotdb.db.metadata.MetadataConstant.ALL_MATCH_PATTERN;
@@ -159,12 +160,17 @@ public class ClusterSchemaTree implements ISchemaTree {
appendSingleMeasurement(
measurementPath,
(MeasurementSchema) measurementPath.getMeasurementSchema(),
+ measurementPath.getTagMap(),
measurementPath.isMeasurementAliasExists() ? measurementPath.getMeasurementAlias() : null,
measurementPath.isUnderAlignedEntity());
}
public void appendSingleMeasurement(
- PartialPath path, MeasurementSchema schema, String alias, boolean isAligned) {
+ PartialPath path,
+ MeasurementSchema schema,
+ Map<String, String> tagMap,
+ String alias,
+ boolean isAligned) {
String[] nodes = path.getNodes();
SchemaNode cur = root;
SchemaNode child;
@@ -177,6 +183,7 @@ public class ClusterSchemaTree implements ISchemaTree {
measurementNode.setAlias(alias);
cur.getAsEntityNode().addAliasChild(alias, measurementNode);
}
+ measurementNode.setTagMap(tagMap);
child = measurementNode;
} else if (i == nodes.length - 2) {
SchemaEntityNode entityNode = new SchemaEntityNode(nodes[i]);
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/common/schematree/node/SchemaMeasurementNode.java b/server/src/main/java/org/apache/iotdb/db/mpp/common/schematree/node/SchemaMeasurementNode.java
index 8a2bdb5c82..d7751ef317 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/common/schematree/node/SchemaMeasurementNode.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/common/schematree/node/SchemaMeasurementNode.java
@@ -25,12 +25,15 @@ import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
+import java.util.Map;
public class SchemaMeasurementNode extends SchemaNode {
private String alias;
private MeasurementSchema schema;
+ private Map<String, String> tagMap;
+
public SchemaMeasurementNode(String name, MeasurementSchema schema) {
super(name);
this.schema = schema;
@@ -48,6 +51,10 @@ public class SchemaMeasurementNode extends SchemaNode {
return schema;
}
+ public Map<String, String> getTagMap() {
+ return tagMap;
+ }
+
@Override
public void replaceChild(String name, SchemaNode newChild) {
throw new UnsupportedOperationException(
@@ -68,6 +75,10 @@ public class SchemaMeasurementNode extends SchemaNode {
this.schema = schema;
}
+ public void setTagMap(Map<String, String> tagMap) {
+ this.tagMap = tagMap;
+ }
+
@Override
public boolean isMeasurement() {
return true;
@@ -90,15 +101,18 @@ public class SchemaMeasurementNode extends SchemaNode {
ReadWriteIOUtils.write(alias, outputStream);
schema.serializeTo(outputStream);
+ ReadWriteIOUtils.write(tagMap, outputStream);
}
public static SchemaMeasurementNode deserialize(InputStream inputStream) throws IOException {
String name = ReadWriteIOUtils.readString(inputStream);
String alias = ReadWriteIOUtils.readString(inputStream);
MeasurementSchema schema = MeasurementSchema.deserializeFrom(inputStream);
+ Map<String, String> tagMap = ReadWriteIOUtils.readMap(inputStream);
SchemaMeasurementNode measurementNode = new SchemaMeasurementNode(name, schema);
measurementNode.setAlias(alias);
+ measurementNode.setTagMap(tagMap);
return measurementNode;
}
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/common/schematree/visitor/SchemaTreeMeasurementVisitor.java b/server/src/main/java/org/apache/iotdb/db/mpp/common/schematree/visitor/SchemaTreeMeasurementVisitor.java
index f86a22ccb3..baddd516f2 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/common/schematree/visitor/SchemaTreeMeasurementVisitor.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/common/schematree/visitor/SchemaTreeMeasurementVisitor.java
@@ -74,6 +74,7 @@ public class SchemaTreeMeasurementVisitor extends SchemaTreeVisitor<MeasurementP
new MeasurementPath(
generateFullPathNodes(nextMatchedNode),
nextMatchedNode.getAsMeasurementNode().getSchema());
+ result.setTagMap(nextMatchedNode.getAsMeasurementNode().getTagMap());
result.setUnderAlignedEntity(ancestorStack.peek().getNode().getAsEntityNode().isAligned());
String alias = nextMatchedNode.getAsMeasurementNode().getAlias();
if (nodes[nodes.length - 1].equals(alias)) {
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/TagAggregationOperator.java b/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/TagAggregationOperator.java
new file mode 100644
index 0000000000..b8fa5b562f
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/TagAggregationOperator.java
@@ -0,0 +1,237 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.mpp.execution.operator.process;
+
+import org.apache.iotdb.db.mpp.aggregation.Aggregator;
+import org.apache.iotdb.db.mpp.execution.operator.Operator;
+import org.apache.iotdb.db.mpp.execution.operator.OperatorContext;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.read.common.block.TsBlock;
+import org.apache.iotdb.tsfile.read.common.block.TsBlockBuilder;
+import org.apache.iotdb.tsfile.read.common.block.column.ColumnBuilder;
+import org.apache.iotdb.tsfile.read.common.block.column.TimeColumnBuilder;
+import org.apache.iotdb.tsfile.utils.Binary;
+import org.apache.iotdb.tsfile.utils.BitMap;
+
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListenableFuture;
+import org.apache.commons.lang3.Validate;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+
+public class TagAggregationOperator implements ProcessOperator {
+
+ private final OperatorContext operatorContext;
+ private final List<List<String>> groups;
+ private final List<List<Aggregator>> groupedAggregators;
+ private final List<Operator> children;
+ private final TsBlock[] inputTsBlocks;
+
+ // This fields record the to be consumed index of each tsBlock.
+ private final int[] consumedIndices;
+ private final BitMap dataReady;
+ private final BitMap childFinished;
+ private final TsBlockBuilder tsBlockBuilder;
+ private final long maxRetainedSize;
+ private final long childrenRetainedSize;
+ private final long maxReturnSize;
+
+ public TagAggregationOperator(
+ OperatorContext operatorContext,
+ List<List<String>> groups,
+ List<List<Aggregator>> groupedAggregators,
+ List<Operator> children,
+ long maxReturnSize) {
+ this.operatorContext = Validate.notNull(operatorContext);
+ this.groups = Validate.notNull(groups);
+ this.groupedAggregators = Validate.notNull(groupedAggregators);
+ this.children = Validate.notNull(children);
+ List<TSDataType> actualOutputColumnTypes = new ArrayList<>();
+ for (String ignored : groups.get(0)) {
+ actualOutputColumnTypes.add(TSDataType.TEXT);
+ }
+ for (int outputColumnIdx = 0;
+ outputColumnIdx < groupedAggregators.get(0).size();
+ outputColumnIdx++) {
+ for (List<Aggregator> aggregators : groupedAggregators) {
+ Aggregator aggregator = aggregators.get(outputColumnIdx);
+ if (aggregator != null) {
+ actualOutputColumnTypes.addAll(Arrays.asList(aggregator.getOutputType()));
+ break;
+ }
+ }
+ }
+ this.tsBlockBuilder = new TsBlockBuilder(actualOutputColumnTypes);
+ // Initialize input tsblocks for each aggregator group.
+ this.inputTsBlocks = new TsBlock[children.size()];
+ this.consumedIndices = new int[children.size()];
+ this.dataReady = new BitMap(children.size());
+ this.childFinished = new BitMap(children.size());
+ this.maxRetainedSize = children.stream().mapToLong(Operator::calculateMaxReturnSize).sum();
+ this.childrenRetainedSize =
+ children.stream().mapToLong(Operator::calculateRetainedSizeAfterCallingNext).sum();
+ this.maxReturnSize = maxReturnSize;
+ }
+
+ @Override
+ public OperatorContext getOperatorContext() {
+ return operatorContext;
+ }
+
+ @Override
+ public TsBlock next() {
+ long maxRuntime = operatorContext.getMaxRunTime().roundTo(TimeUnit.NANOSECONDS);
+ long start = System.nanoTime();
+ boolean successful = true;
+ while (System.nanoTime() - start < maxRuntime && !tsBlockBuilder.isFull() && successful) {
+ successful = processOneRow();
+ }
+ TsBlock tsBlock = null;
+ if (tsBlockBuilder.getPositionCount() > 0) {
+ tsBlock = tsBlockBuilder.build();
+ }
+ tsBlockBuilder.reset();
+ return tsBlock;
+ }
+
+ private boolean processOneRow() {
+ for (int i = 0; i < children.size(); i++) {
+ if (dataReady.isMarked(i)) {
+ continue;
+ }
+ if (dataUnavailable(i)) {
+ // Find next non-empty tsblock of the child
+ do {
+ inputTsBlocks[i] = children.get(i).next();
+ } while (inputTsBlocks[i] != null && inputTsBlocks[i].isEmpty());
+ consumedIndices[i] = 0;
+ }
+
+ // Blocked by children i
+ if (inputTsBlocks[i] == null) {
+ return false;
+ }
+ dataReady.mark(i);
+ }
+
+ TsBlock[] rowBlocks = new TsBlock[children.size()];
+ for (int i = 0; i < children.size(); i++) {
+ rowBlocks[i] = inputTsBlocks[i].getRegion(consumedIndices[i], 1);
+ }
+ for (int groupIdx = 0; groupIdx < groups.size(); groupIdx++) {
+ List<String> group = groups.get(groupIdx);
+ List<Aggregator> aggregators = groupedAggregators.get(groupIdx);
+
+ for (Aggregator aggregator : aggregators) {
+ if (aggregator == null) {
+ continue;
+ }
+ aggregator.reset();
+ aggregator.processTsBlocks(rowBlocks);
+ }
+
+ TimeColumnBuilder timeColumnBuilder = tsBlockBuilder.getTimeColumnBuilder();
+ timeColumnBuilder.writeLong(rowBlocks[0].getStartTime());
+ ColumnBuilder[] columnBuilders = tsBlockBuilder.getValueColumnBuilders();
+
+ for (int i = 0; i < group.size(); i++) {
+ if (group.get(i) == null) {
+ columnBuilders[i].writeBinary(new Binary("NULL"));
+ } else {
+ columnBuilders[i].writeBinary(new Binary(group.get(i)));
+ }
+ }
+ for (int i = 0; i < aggregators.size(); i++) {
+ Aggregator aggregator = aggregators.get(i);
+ ColumnBuilder columnBuilder = columnBuilders[i + group.size()];
+ if (aggregator == null) {
+ columnBuilder.appendNull();
+ } else {
+ aggregator.outputResult(new ColumnBuilder[] {columnBuilder});
+ }
+ }
+ tsBlockBuilder.declarePosition();
+ }
+
+ // Reset dataReady for next iteration
+ for (int i = 0; i < children.size(); i++) {
+ consumedIndices[i]++;
+ dataReady.unmark(i);
+ }
+ return true;
+ }
+
+ @Override
+ public boolean hasNext() {
+ for (int i = 0; i < children.size(); i++) {
+ if (dataUnavailable(i) && !children.get(i).hasNext()) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ @Override
+ public boolean isFinished() {
+ return !this.hasNext();
+ }
+
+ @Override
+ public ListenableFuture<?> isBlocked() {
+ List<ListenableFuture<?>> listenableFutures = new ArrayList<>();
+ for (int i = 0; i < children.size(); i++) {
+ ListenableFuture<?> blocked = children.get(i).isBlocked();
+ if (!blocked.isDone() && consumedIndices[i] == inputTsBlocks[i].getPositionCount()) {
+ listenableFutures.add(blocked);
+ }
+ }
+ return listenableFutures.isEmpty() ? NOT_BLOCKED : Futures.successfulAsList(listenableFutures);
+ }
+
+ @Override
+ public long calculateMaxPeekMemory() {
+ return maxReturnSize + maxRetainedSize + childrenRetainedSize;
+ }
+
+ @Override
+ public long calculateMaxReturnSize() {
+ return maxReturnSize;
+ }
+
+ @Override
+ public long calculateRetainedSizeAfterCallingNext() {
+ return maxRetainedSize + childrenRetainedSize;
+ }
+
+ @Override
+ public void close() throws Exception {
+ for (Operator child : children) {
+ child.close();
+ }
+ }
+
+ private boolean dataUnavailable(int index) {
+ return inputTsBlocks[index] == null
+ || consumedIndices[index] == inputTsBlocks[index].getPositionCount();
+ }
+}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/schema/SchemaFetchScanOperator.java b/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/schema/SchemaFetchScanOperator.java
index 4e07c4178c..a0af72b3b0 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/schema/SchemaFetchScanOperator.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/schema/SchemaFetchScanOperator.java
@@ -56,6 +56,7 @@ public class SchemaFetchScanOperator implements SourceOperator {
private final Map<Integer, Template> templateMap;
private final ISchemaRegion schemaRegion;
+ private final boolean withTags;
private boolean isFinished = false;
@@ -64,12 +65,14 @@ public class SchemaFetchScanOperator implements SourceOperator {
OperatorContext context,
PathPatternTree patternTree,
Map<Integer, Template> templateMap,
- ISchemaRegion schemaRegion) {
+ ISchemaRegion schemaRegion,
+ boolean withTags) {
this.sourceId = planNodeId;
this.operatorContext = context;
this.patternTree = patternTree;
this.schemaRegion = schemaRegion;
this.templateMap = templateMap;
+ this.withTags = withTags;
}
@Override
@@ -110,7 +113,7 @@ public class SchemaFetchScanOperator implements SourceOperator {
ClusterSchemaTree schemaTree = new ClusterSchemaTree();
List<PartialPath> partialPathList = patternTree.getAllPathPatterns();
for (PartialPath path : partialPathList) {
- schemaTree.appendMeasurementPaths(schemaRegion.fetchSchema(path, templateMap));
+ schemaTree.appendMeasurementPaths(schemaRegion.fetchSchema(path, templateMap, withTags));
}
ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/Analysis.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/Analysis.java
index d45d98e22c..57c7f6309d 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/Analysis.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/Analysis.java
@@ -94,6 +94,24 @@ public class Analysis {
// map from raw path to grouped path in `GROUP BY LEVEL` clause
private Map<Expression, Expression> rawPathToGroupedPathMap;
+ // tag keys specified in `GROUP BY TAG` clause
+ private List<String> tagKeys;
+
+ // {tag values -> {grouped expression -> output expressions}}
+ // For different combination of tag keys, the grouped expression may be different. Let's say there
+ // are 3 timeseries root.sg.d1.temperature, root.sg.d1.status, root.sg.d2.temperature, and their
+ // tags are [k1=v1], [k1=v1] and [k1=v2] respectively. For query "SELECT last_value(**) FROM root
+ // GROUP BY k1", timeseries are grouped by their tags into 2 buckets. Bucket [v1] has
+ // [root.sg.d1.temperature, root.sg.d1.status], while bucket [v2] has [root.sg.d2.temperature].
+ // Thus, the aggregation results of bucket [v1] and [v2] are different. Bucket [v1] has 2
+ // aggregation results last_value(temperature) and last_value(status), whereas bucket [v2] only
+ // has [last_value(temperature)].
+ private Map<List<String>, LinkedHashMap<Expression, List<Expression>>>
+ tagValuesToGroupedTimeseriesOperands;
+
+ // All the possible aggregation results.
+ private List<Expression> groupByTagOutputExpressions;
+
private boolean isRawDataSource;
/////////////////////////////////////////////////////////////////////////////////////////////////
@@ -465,4 +483,31 @@ public class Analysis {
public void setOutputExpressions(List<Pair<Expression, String>> outputExpressions) {
this.outputExpressions = outputExpressions;
}
+
+ public List<String> getTagKeys() {
+ return tagKeys;
+ }
+
+ public void setTagKeys(List<String> tagKeys) {
+ this.tagKeys = tagKeys;
+ }
+
+ public Map<List<String>, LinkedHashMap<Expression, List<Expression>>>
+ getTagValuesToGroupedTimeseriesOperands() {
+ return tagValuesToGroupedTimeseriesOperands;
+ }
+
+ public void setTagValuesToGroupedTimeseriesOperands(
+ Map<List<String>, LinkedHashMap<Expression, List<Expression>>>
+ tagValuesToGroupedTimeseriesOperands) {
+ this.tagValuesToGroupedTimeseriesOperands = tagValuesToGroupedTimeseriesOperands;
+ }
+
+ public List<Expression> getGroupByTagOutputExpressions() {
+ return groupByTagOutputExpressions;
+ }
+
+ public void setGroupByTagOutputExpressions(List<Expression> groupByTagOutputExpressions) {
+ this.groupByTagOutputExpressions = groupByTagOutputExpressions;
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/AnalyzeVisitor.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/AnalyzeVisitor.java
index a5dace95a8..2ca6c0712e 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/AnalyzeVisitor.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/AnalyzeVisitor.java
@@ -41,6 +41,7 @@ import org.apache.iotdb.db.mpp.common.schematree.ISchemaTree;
import org.apache.iotdb.db.mpp.common.schematree.PathPatternTree;
import org.apache.iotdb.db.mpp.plan.expression.Expression;
import org.apache.iotdb.db.mpp.plan.expression.ExpressionType;
+import org.apache.iotdb.db.mpp.plan.expression.leaf.ConstantOperand;
import org.apache.iotdb.db.mpp.plan.expression.leaf.TimeSeriesOperand;
import org.apache.iotdb.db.mpp.plan.expression.multi.FunctionExpression;
import org.apache.iotdb.db.mpp.plan.planner.plan.parameter.FillDescriptor;
@@ -50,6 +51,7 @@ import org.apache.iotdb.db.mpp.plan.statement.Statement;
import org.apache.iotdb.db.mpp.plan.statement.StatementNode;
import org.apache.iotdb.db.mpp.plan.statement.StatementVisitor;
import org.apache.iotdb.db.mpp.plan.statement.component.FillComponent;
+import org.apache.iotdb.db.mpp.plan.statement.component.GroupByTagComponent;
import org.apache.iotdb.db.mpp.plan.statement.component.GroupByTimeComponent;
import org.apache.iotdb.db.mpp.plan.statement.component.Ordering;
import org.apache.iotdb.db.mpp.plan.statement.component.ResultColumn;
@@ -170,7 +172,12 @@ public class AnalyzeVisitor extends StatementVisitor<Analysis, MPPQueryContext>
// request schema fetch API
logger.info("{} fetch query schema...", getLogHeader());
- ISchemaTree schemaTree = schemaFetcher.fetchSchema(patternTree);
+ ISchemaTree schemaTree;
+ if (queryStatement.isGroupByTag()) {
+ schemaTree = schemaFetcher.fetchSchemaWithTags(patternTree);
+ } else {
+ schemaTree = schemaFetcher.fetchSchema(patternTree);
+ }
logger.info("{} fetch schema done", getLogHeader());
// If there is no leaf node in the schema tree, the query should be completed immediately
if (schemaTree.isEmpty()) {
@@ -417,6 +424,25 @@ public class AnalyzeVisitor extends StatementVisitor<Analysis, MPPQueryContext>
analysis.setRawPathToGroupedPathMap(rawPathToGroupedPathMap);
}
+ if (queryStatement.isGroupByTag()) {
+ if (hasValueFilter) {
+ throw new SemanticException("Only time filters are supported in GROUP BY TAGS query");
+ }
+ Map<List<String>, LinkedHashMap<Expression, List<Expression>>>
+ tagValuesToGroupedTimeseriesOperands = new HashMap<>();
+ List<Expression> groupByTagOutputExpressions = new ArrayList<>();
+ analyzeGroupByTag(
+ analysis,
+ queryStatement.getGroupByTagComponent(),
+ outputExpressions,
+ tagValuesToGroupedTimeseriesOperands,
+ groupByTagOutputExpressions,
+ schemaTree.getAllMeasurement());
+ analysis.setTagKeys(queryStatement.getGroupByTagComponent().getTagKeys());
+ analysis.setTagValuesToGroupedTimeseriesOperands(tagValuesToGroupedTimeseriesOperands);
+ analysis.setGroupByTagOutputExpressions(groupByTagOutputExpressions);
+ }
+
// true if nested expressions and UDFs exist in aggregation function
// i.e. select sum(s1 + 1) from root.sg.d1
boolean isHasRawDataInputAggregation = false;
@@ -452,6 +478,8 @@ public class AnalyzeVisitor extends StatementVisitor<Analysis, MPPQueryContext>
// generate sourceExpression according to transformExpressions
Set<Expression> sourceExpressions = new HashSet<>();
boolean isValueFilterAggregation = queryStatement.isAggregationQuery() && hasValueFilter;
+ // isRawDataSource means we have to read the raw data for the result. The result cannot be
+ // calculated directly from the statistics.
boolean isRawDataSource =
!queryStatement.isAggregationQuery()
|| isValueFilterAggregation
@@ -711,6 +739,65 @@ public class AnalyzeVisitor extends StatementVisitor<Analysis, MPPQueryContext>
return new Pair<>(globalTimeFilter, hasValueFilter);
}
+ private void analyzeGroupByTag(
+ Analysis analysis,
+ GroupByTagComponent groupByTagComponent,
+ List<Pair<Expression, String>> outputExpressions,
+ Map<List<String>, LinkedHashMap<Expression, List<Expression>>>
+ tagValuesToGroupedTimeseriesOperands,
+ List<Expression> groupByTagOutputExpressions,
+ List<MeasurementPath> allSelectedPath) {
+ List<String> tagKeys = groupByTagComponent.getTagKeys();
+ Map<MeasurementPath, Map<String, String>> queriedTagMap = new HashMap<>();
+ allSelectedPath.forEach(v -> queriedTagMap.put(v, v.getTagMap()));
+ for (Pair<Expression, String> outputExpressionAndAlias : outputExpressions) {
+ if (!(outputExpressionAndAlias.getLeft() instanceof FunctionExpression
+ && outputExpressionAndAlias.getLeft().getExpressions().get(0) instanceof TimeSeriesOperand
+ && outputExpressionAndAlias.getLeft().isBuiltInAggregationFunctionExpression())) {
+ throw new SemanticException(
+ outputExpressionAndAlias.getLeft()
+ + " can't be used in group by tag. It will be supported in the future.");
+ }
+ FunctionExpression outputExpression = (FunctionExpression) outputExpressionAndAlias.getLeft();
+ MeasurementPath measurementPath =
+ (MeasurementPath)
+ ((TimeSeriesOperand) outputExpression.getExpressions().get(0)).getPath();
+ Expression measurementExpression =
+ new ConstantOperand(measurementPath.getSeriesType(), measurementPath.getMeasurement());
+ Expression groupedExpression =
+ new FunctionExpression(
+ outputExpression.getFunctionName(),
+ outputExpression.getFunctionAttributes(),
+ Collections.singletonList(measurementExpression));
+ if (!groupByTagOutputExpressions.contains(groupedExpression)) {
+ groupByTagOutputExpressions.add(groupedExpression);
+ }
+ Map<String, String> tagMap = queriedTagMap.get(measurementPath);
+ List<String> tagValues = new ArrayList<>();
+ for (String tagKey : tagKeys) {
+ tagValues.add(tagMap.get(tagKey));
+ }
+ tagValuesToGroupedTimeseriesOperands
+ .computeIfAbsent(tagValues, key -> new LinkedHashMap<>())
+ .computeIfAbsent(groupedExpression, key -> new ArrayList<>())
+ .add(outputExpression.getExpressions().get(0));
+ }
+
+ outputExpressions.clear();
+ for (String tagKey : tagKeys) {
+ Expression tagKeyExpression = new ConstantOperand(TSDataType.TEXT, tagKey);
+ analyzeExpression(analysis, tagKeyExpression);
+ outputExpressions.add(new Pair<>(tagKeyExpression, null));
+ }
+ for (Expression groupByTagOutputExpression : groupByTagOutputExpressions) {
+ // TODO: support alias
+ ConstantOperand measurementExpression =
+ (ConstantOperand) groupByTagOutputExpression.getExpressions().get(0);
+ analyzeExpression(analysis, measurementExpression);
+ outputExpressions.add(new Pair<>(groupByTagOutputExpression, null));
+ }
+ }
+
private void updateSource(
Expression selectExpr, Set<Expression> sourceExpressions, boolean isRawDataSource) {
sourceExpressions.addAll(
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/ClusterSchemaFetcher.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/ClusterSchemaFetcher.java
index 77ff8e9abb..fe8a756818 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/ClusterSchemaFetcher.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/ClusterSchemaFetcher.java
@@ -96,12 +96,21 @@ public class ClusterSchemaFetcher implements ISchemaFetcher {
@Override
public ClusterSchemaTree fetchSchema(PathPatternTree patternTree) {
+ return fetchSchema(patternTree, false);
+ }
+
+ @Override
+ public ClusterSchemaTree fetchSchemaWithTags(PathPatternTree patternTree) {
+ return fetchSchema(patternTree, true);
+ }
+
+ private ClusterSchemaTree fetchSchema(PathPatternTree patternTree, boolean withTags) {
Map<Integer, Template> templateMap = new HashMap<>();
patternTree.constructTree();
for (PartialPath pattern : patternTree.getAllPathPatterns()) {
templateMap.putAll(templateManager.checkAllRelatedTemplate(pattern));
}
- return executeSchemaFetchQuery(new SchemaFetchStatement(patternTree, templateMap));
+ return executeSchemaFetchQuery(new SchemaFetchStatement(patternTree, templateMap, withTags));
}
private ClusterSchemaTree executeSchemaFetchQuery(SchemaFetchStatement schemaFetchStatement) {
@@ -336,6 +345,7 @@ public class ClusterSchemaFetcher implements ISchemaFetcher {
devicePath.concatNode(entry.getKey()),
(MeasurementSchema) entry.getValue(),
null,
+ null,
template.isDirectAligned());
}
@@ -416,6 +426,7 @@ public class ClusterSchemaFetcher implements ISchemaFetcher {
new MeasurementSchema(
measurements.get(i), tsDataTypes.get(i), encodings.get(i), compressors.get(i)),
null,
+ null,
isAligned);
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/ISchemaFetcher.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/ISchemaFetcher.java
index e175f59174..218b8bb9fa 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/ISchemaFetcher.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/ISchemaFetcher.java
@@ -37,6 +37,8 @@ public interface ISchemaFetcher {
ISchemaTree fetchSchema(PathPatternTree patternTree);
+ ISchemaTree fetchSchemaWithTags(PathPatternTree patternTree);
+
ISchemaTree fetchSchemaWithAutoCreate(
PartialPath devicePath,
String[] measurements,
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/StandaloneSchemaFetcher.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/StandaloneSchemaFetcher.java
index 4e9cc1cc67..c37e4b6e68 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/StandaloneSchemaFetcher.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/StandaloneSchemaFetcher.java
@@ -58,6 +58,15 @@ public class StandaloneSchemaFetcher implements ISchemaFetcher {
@Override
public ClusterSchemaTree fetchSchema(PathPatternTree patternTree) {
+ return fetchSchema(patternTree, false);
+ }
+
+ @Override
+ public ClusterSchemaTree fetchSchemaWithTags(PathPatternTree patternTree) {
+ return fetchSchema(patternTree, true);
+ }
+
+ private ClusterSchemaTree fetchSchema(PathPatternTree patternTree, boolean withTags) {
patternTree.constructTree();
Set<String> storageGroupSet = new HashSet<>();
ClusterSchemaTree schemaTree = new ClusterSchemaTree();
@@ -70,7 +79,8 @@ public class StandaloneSchemaFetcher implements ISchemaFetcher {
SchemaRegionId schemaRegionId =
localConfigNode.getBelongedSchemaRegionId(storageGroupPath);
ISchemaRegion schemaRegion = schemaEngine.getSchemaRegion(schemaRegionId);
- schemaTree.appendMeasurementPaths(schemaRegion.getMeasurementPaths(pathPattern, false));
+ schemaTree.appendMeasurementPaths(
+ schemaRegion.getMeasurementPaths(pathPattern, false, withTags));
}
}
} catch (MetadataException e) {
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/parser/ASTVisitor.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/parser/ASTVisitor.java
index 6420fc7999..aeaab64353 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/parser/ASTVisitor.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/parser/ASTVisitor.java
@@ -69,6 +69,7 @@ import org.apache.iotdb.db.mpp.plan.statement.component.FillComponent;
import org.apache.iotdb.db.mpp.plan.statement.component.FillPolicy;
import org.apache.iotdb.db.mpp.plan.statement.component.FromComponent;
import org.apache.iotdb.db.mpp.plan.statement.component.GroupByLevelComponent;
+import org.apache.iotdb.db.mpp.plan.statement.component.GroupByTagComponent;
import org.apache.iotdb.db.mpp.plan.statement.component.GroupByTimeComponent;
import org.apache.iotdb.db.mpp.plan.statement.component.HavingCondition;
import org.apache.iotdb.db.mpp.plan.statement.component.OrderByComponent;
@@ -150,6 +151,9 @@ import org.apache.iotdb.db.qp.sql.IoTDBSqlParser.CountTimeseriesContext;
import org.apache.iotdb.db.qp.sql.IoTDBSqlParser.CreateFunctionContext;
import org.apache.iotdb.db.qp.sql.IoTDBSqlParser.DropFunctionContext;
import org.apache.iotdb.db.qp.sql.IoTDBSqlParser.ExpressionContext;
+import org.apache.iotdb.db.qp.sql.IoTDBSqlParser.GroupByTagClauseContext;
+import org.apache.iotdb.db.qp.sql.IoTDBSqlParser.GroupByTagStatementContext;
+import org.apache.iotdb.db.qp.sql.IoTDBSqlParser.IdentifierContext;
import org.apache.iotdb.db.qp.sql.IoTDBSqlParser.ShowFunctionsContext;
import org.apache.iotdb.db.qp.sql.IoTDBSqlParser.UriContext;
import org.apache.iotdb.db.qp.sql.IoTDBSqlParserBaseVisitor;
@@ -968,6 +972,14 @@ public class ASTVisitor extends IoTDBSqlParserBaseVisitor<Statement> {
queryStatement.setGroupByLevelComponent(groupByLevelComponent);
}
+ if (ctx.TAGS() != null) {
+ List<String> tagKeys = new ArrayList<>();
+ for (IdentifierContext identifierContext : ctx.identifier()) {
+ tagKeys.add(identifierContext.getText());
+ }
+ queryStatement.setGroupByTagComponent(new GroupByTagComponent(tagKeys));
+ }
+
// parse fill clause
if (ctx.fillClause() != null) {
parseFillClause(ctx.fillClause());
@@ -1089,6 +1101,31 @@ public class ASTVisitor extends IoTDBSqlParserBaseVisitor<Statement> {
queryStatement.setHavingCondition(new HavingCondition(predicate));
}
+ @Override
+ public Statement visitGroupByTagStatement(GroupByTagStatementContext ctx) {
+ // parse group by tag clause
+ parseGroupByTagClause(ctx.groupByTagClause());
+
+ // parse order by time
+ if (ctx.orderByClause() != null) {
+ parseOrderByClause(ctx.orderByClause());
+ }
+
+ return queryStatement;
+ }
+
+ public void parseGroupByTagClause(GroupByTagClauseContext ctx) {
+ Set<String> tagKeys = new LinkedHashSet<>();
+ for (IdentifierContext identifierContext : ctx.identifier()) {
+ String key = parseIdentifier(identifierContext.getText());
+ if (tagKeys.contains(key)) {
+ throw new SemanticException("duplicated key in GROUP BY TAGS: " + key);
+ }
+ tagKeys.add(key);
+ }
+ queryStatement.setGroupByTagComponent(new GroupByTagComponent(new ArrayList<>(tagKeys)));
+ }
+
// Fill Clause
@Override
public Statement visitFillStatement(IoTDBSqlParser.FillStatementContext ctx) {
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/LogicalPlanBuilder.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/LogicalPlanBuilder.java
index 3a35c3369d..18f73a4727 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/LogicalPlanBuilder.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/LogicalPlanBuilder.java
@@ -56,6 +56,8 @@ import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.DeviceViewNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.FillNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.FilterNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.GroupByLevelNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.GroupByTagNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.GroupByTagNode.GroupByTagAggregationDescriptor;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.LimitNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.OffsetNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.SlidingWindowAggregationNode;
@@ -93,6 +95,8 @@ import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
+import java.util.Iterator;
+import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
@@ -205,7 +209,12 @@ public class LogicalPlanBuilder {
GroupByTimeParameter groupByTimeParameter,
Set<Expression> aggregationExpressions,
Set<Expression> aggregationTransformExpressions,
- Map<Expression, Set<Expression>> groupByLevelExpressions) {
+ Map<Expression, Set<Expression>> groupByLevelExpressions,
+ List<String> tagKeys,
+ Map<List<String>, LinkedHashMap<Expression, List<Expression>>>
+ tagValuesToGroupedTimeseriesOperands,
+ List<Expression> groupByTagOutputExpressions) {
+
boolean needCheckAscending = groupByTimeParameter == null;
Map<PartialPath, List<AggregationDescriptor>> ascendingAggregations = new HashMap<>();
Map<PartialPath, List<AggregationDescriptor>> descendingAggregations = new HashMap<>();
@@ -235,7 +244,10 @@ public class LogicalPlanBuilder {
scanOrder,
groupByTimeParameter,
aggregationExpressions,
- groupByLevelExpressions);
+ groupByLevelExpressions,
+ tagKeys,
+ tagValuesToGroupedTimeseriesOperands,
+ groupByTagOutputExpressions);
}
public LogicalPlanBuilder planAggregationSourceWithIndexAdjust(
@@ -300,7 +312,10 @@ public class LogicalPlanBuilder {
scanOrder,
groupByTimeParameter,
aggregationExpressions,
- groupByLevelExpressions);
+ groupByLevelExpressions,
+ null,
+ null,
+ null);
}
private AggregationDescriptor createAggregationDescriptor(
@@ -376,7 +391,11 @@ public class LogicalPlanBuilder {
Ordering scanOrder,
GroupByTimeParameter groupByTimeParameter,
Set<Expression> aggregationExpressions,
- Map<Expression, Set<Expression>> groupByLevelExpressions) {
+ Map<Expression, Set<Expression>> groupByLevelExpressions,
+ List<String> tagKeys,
+ Map<List<String>, LinkedHashMap<Expression, List<Expression>>>
+ tagValuesToGroupedTimeseriesOperands,
+ List<Expression> groupByTagOutputExpressions) {
if (curStep.isOutputPartial()) {
if (groupByTimeParameter != null && groupByTimeParameter.hasOverlap()) {
curStep =
@@ -409,6 +428,18 @@ public class LogicalPlanBuilder {
groupByTimeParameter,
scanOrder);
}
+ if (tagKeys != null) {
+ curStep = AggregationStep.FINAL;
+ this.root =
+ createGroupByTagNode(
+ tagKeys,
+ tagValuesToGroupedTimeseriesOperands,
+ groupByTagOutputExpressions,
+ sourceNodeList,
+ curStep,
+ groupByTimeParameter,
+ scanOrder);
+ }
}
} else {
this.root = convergeWithTimeJoin(sourceNodeList, scanOrder);
@@ -596,6 +627,54 @@ public class LogicalPlanBuilder {
scanOrder);
}
+ private PlanNode createGroupByTagNode(
+ List<String> tagKeys,
+ Map<List<String>, LinkedHashMap<Expression, List<Expression>>>
+ tagValuesToGroupedTimeseriesOperands,
+ List<Expression> groupByTagOutputExpressions,
+ List<PlanNode> children,
+ AggregationStep curStep,
+ GroupByTimeParameter groupByTimeParameter,
+ Ordering scanOrder) {
+ Map<List<String>, List<GroupByTagAggregationDescriptor>> tagValuesToAggregationDescriptors =
+ new HashMap<>();
+ for (List<String> tagValues : tagValuesToGroupedTimeseriesOperands.keySet()) {
+ LinkedHashMap<Expression, List<Expression>> groupedTimeseriesOperands =
+ tagValuesToGroupedTimeseriesOperands.get(tagValues);
+ List<GroupByTagAggregationDescriptor> aggregationDescriptors = new ArrayList<>();
+
+ Iterator<Expression> iter = groupedTimeseriesOperands.keySet().iterator();
+ for (Expression groupByTagOutputExpression : groupByTagOutputExpressions) {
+ if (!iter.hasNext()) {
+ aggregationDescriptors.add(null);
+ continue;
+ }
+ Expression next = iter.next();
+ if (next.equals(groupByTagOutputExpression)) {
+ String functionName = ((FunctionExpression) next).getFunctionName().toUpperCase();
+ GroupByTagAggregationDescriptor aggregationDescriptor =
+ new GroupByTagAggregationDescriptor(
+ functionName, curStep, groupedTimeseriesOperands.get(next), next);
+ aggregationDescriptors.add(aggregationDescriptor);
+ } else {
+ aggregationDescriptors.add(null);
+ }
+ }
+ tagValuesToAggregationDescriptors.put(tagValues, aggregationDescriptors);
+ }
+
+ return new GroupByTagNode(
+ context.getQueryId().genPlanNodeId(),
+ children,
+ groupByTimeParameter,
+ scanOrder,
+ tagKeys,
+ tagValuesToAggregationDescriptors,
+ groupByTagOutputExpressions.stream()
+ .map(Expression::toString)
+ .collect(Collectors.toList()));
+ }
+
private SeriesAggregationSourceNode createAggregationScanNode(
PartialPath selectPath,
List<AggregationDescriptor> aggregationDescriptorList,
@@ -780,7 +859,8 @@ public class LogicalPlanBuilder {
public LogicalPlanBuilder planSchemaFetchSource(
List<String> storageGroupList,
PathPatternTree patternTree,
- Map<Integer, Template> templateMap) {
+ Map<Integer, Template> templateMap,
+ boolean withTags) {
PartialPath storageGroupPath;
for (String storageGroup : storageGroupList) {
try {
@@ -796,7 +876,8 @@ public class LogicalPlanBuilder {
context.getQueryId().genPlanNodeId(),
storageGroupPath,
overlappedPatternTree,
- templateMap));
+ templateMap,
+ withTags));
} catch (IllegalPathException e) {
// definitely won't happen
throw new RuntimeException(e);
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/LogicalPlanVisitor.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/LogicalPlanVisitor.java
index b59e490ebe..ff5c6cf0dc 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/LogicalPlanVisitor.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/LogicalPlanVisitor.java
@@ -280,6 +280,7 @@ public class LogicalPlanVisitor extends StatementVisitor<PlanNode, MPPQueryConte
} else {
AggregationStep curStep =
(analysis.getGroupByLevelExpressions() != null
+ || analysis.getTagValuesToGroupedTimeseriesOperands() != null
|| (analysis.getGroupByTimeParameter() != null
&& analysis.getGroupByTimeParameter().hasOverlap()))
? AggregationStep.PARTIAL
@@ -332,7 +333,10 @@ public class LogicalPlanVisitor extends StatementVisitor<PlanNode, MPPQueryConte
analysis.getGroupByTimeParameter(),
aggregationExpressions,
aggregationTransformExpressions,
- analysis.getGroupByLevelExpressions());
+ analysis.getGroupByLevelExpressions(),
+ analysis.getTagKeys(),
+ analysis.getTagValuesToGroupedTimeseriesOperands(),
+ analysis.getGroupByTagOutputExpressions());
if (queryStatement.isGroupByLevel()) {
planBuilder = // plan Having with GroupByLevel
@@ -714,7 +718,8 @@ public class LogicalPlanVisitor extends StatementVisitor<PlanNode, MPPQueryConte
.planSchemaFetchSource(
storageGroupList,
schemaFetchStatement.getPatternTree(),
- schemaFetchStatement.getTemplateMap())
+ schemaFetchStatement.getTemplateMap(),
+ schemaFetchStatement.isWithTags())
.getRoot();
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/OperatorTreeGenerator.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/OperatorTreeGenerator.java
index 9c80c7c97e..8a32d088db 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/OperatorTreeGenerator.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/OperatorTreeGenerator.java
@@ -49,6 +49,7 @@ import org.apache.iotdb.db.mpp.execution.operator.process.OffsetOperator;
import org.apache.iotdb.db.mpp.execution.operator.process.ProcessOperator;
import org.apache.iotdb.db.mpp.execution.operator.process.RawDataAggregationOperator;
import org.apache.iotdb.db.mpp.execution.operator.process.SlidingWindowAggregationOperator;
+import org.apache.iotdb.db.mpp.execution.operator.process.TagAggregationOperator;
import org.apache.iotdb.db.mpp.execution.operator.process.TransformOperator;
import org.apache.iotdb.db.mpp.execution.operator.process.fill.IFill;
import org.apache.iotdb.db.mpp.execution.operator.process.fill.ILinearFill;
@@ -135,6 +136,7 @@ import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.ExchangeNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.FillNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.FilterNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.GroupByLevelNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.GroupByTagNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.LimitNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.OffsetNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.SlidingWindowAggregationNode;
@@ -180,6 +182,7 @@ import org.apache.commons.lang3.Validate;
import java.io.IOException;
import java.util.ArrayList;
+import java.util.Collection;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
@@ -1086,6 +1089,70 @@ public class OperatorTreeGenerator extends PlanVisitor<Operator, LocalExecutionP
operatorContext, aggregators, timeRangeIterator, children, maxReturnSize);
}
+ @Override
+ public Operator visitGroupByTag(GroupByTagNode node, LocalExecutionPlanContext context) {
+ checkArgument(node.getTagKeys().size() >= 1, "GroupByTag tag keys cannot be empty");
+ checkArgument(
+ node.getTagValuesToAggregationDescriptors().size() >= 1,
+ "GroupByTag aggregation descriptors cannot be empty");
+
+ List<Operator> children =
+ node.getChildren().stream()
+ .map(child -> child.accept(this, context))
+ .collect(Collectors.toList());
+
+ boolean ascending = node.getScanOrder() == Ordering.ASC;
+ Map<String, List<InputLocation>> layout = makeLayout(node);
+ List<List<String>> groups = new ArrayList<>();
+ List<List<Aggregator>> groupedAggregators = new ArrayList<>();
+ int aggregatorCount = 0;
+ for (Map.Entry<List<String>, List<GroupByTagNode.GroupByTagAggregationDescriptor>> entry :
+ node.getTagValuesToAggregationDescriptors().entrySet()) {
+ groups.add(entry.getKey());
+ List<Aggregator> aggregators = new ArrayList<>();
+ for (GroupByTagNode.GroupByTagAggregationDescriptor aggregationDescriptor :
+ entry.getValue()) {
+ if (aggregationDescriptor == null) {
+ aggregators.add(null);
+ continue;
+ }
+ List<InputLocation[]> inputLocations = calcInputLocationList(aggregationDescriptor, layout);
+ TSDataType seriesDataType =
+ context
+ .getTypeProvider()
+ .getType(aggregationDescriptor.getInputExpressions().get(0).getExpressionString());
+ aggregators.add(
+ new Aggregator(
+ AccumulatorFactory.createAccumulator(
+ aggregationDescriptor.getAggregationType(), seriesDataType, ascending),
+ aggregationDescriptor.getStep(),
+ inputLocations));
+ }
+ groupedAggregators.add(aggregators);
+ aggregatorCount += aggregators.size();
+ }
+ GroupByTimeParameter groupByTimeParameter = node.getGroupByTimeParameter();
+ ITimeRangeIterator timeRangeIterator =
+ initTimeRangeIterator(groupByTimeParameter, ascending, false);
+ List<AggregationDescriptor> aggregationDescriptors =
+ node.getTagValuesToAggregationDescriptors().values().stream()
+ .flatMap(Collection::stream)
+ .collect(Collectors.toList());
+ long maxReturnSize =
+ calculateMaxAggregationResultSize(
+ aggregationDescriptors, timeRangeIterator, context.getTypeProvider());
+ OperatorContext operatorContext =
+ context
+ .getInstanceContext()
+ .addOperatorContext(
+ context.getNextOperatorId(),
+ node.getPlanNodeId(),
+ TagAggregationOperator.class.getSimpleName());
+ context.getTimeSliceAllocator().recordExecutionWeight(operatorContext, aggregatorCount);
+ return new TagAggregationOperator(
+ operatorContext, groups, groupedAggregators, children, maxReturnSize);
+ }
+
@Override
public Operator visitSlidingWindowAggregation(
SlidingWindowAggregationNode node, LocalExecutionPlanContext context) {
@@ -1417,8 +1484,8 @@ public class OperatorTreeGenerator extends PlanVisitor<Operator, LocalExecutionP
operatorContext,
node.getPatternTree(),
node.getTemplateMap(),
- ((SchemaDriverContext) (context.getInstanceContext().getDriverContext()))
- .getSchemaRegion());
+ ((SchemaDriverContext) (context.getInstanceContext().getDriverContext())).getSchemaRegion(),
+ node.isWithTags());
}
@Override
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/distribution/ExchangeNodeAdder.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/distribution/ExchangeNodeAdder.java
index d21bf1a195..37c487b4ab 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/distribution/ExchangeNodeAdder.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/distribution/ExchangeNodeAdder.java
@@ -37,6 +37,7 @@ import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.DeviceViewNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.ExchangeNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.FilterNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.GroupByLevelNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.GroupByTagNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.MultiChildNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.SlidingWindowAggregationNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.TimeJoinNode;
@@ -236,6 +237,10 @@ public class ExchangeNodeAdder extends PlanVisitor<PlanNode, NodeGroupContext> {
return processOneChildNode(node, context);
}
+ public PlanNode visitGroupByTag(GroupByTagNode node, NodeGroupContext context) {
+ return processMultiChildNode(node, context);
+ }
+
private PlanNode processMultiChildNode(MultiChildNode node, NodeGroupContext context) {
MultiChildNode newNode = (MultiChildNode) node.clone();
List<PlanNode> visitedChildren = new ArrayList<>();
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/distribution/SourceRewriter.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/distribution/SourceRewriter.java
index c608e09e1b..4f4b8dd422 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/distribution/SourceRewriter.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/distribution/SourceRewriter.java
@@ -37,6 +37,7 @@ import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.AggregationNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.DeviceMergeNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.DeviceViewNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.GroupByLevelNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.GroupByTagNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.MultiChildNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.SlidingWindowAggregationNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.TimeJoinNode;
@@ -64,6 +65,7 @@ import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
+import java.util.Map.Entry;
import java.util.Set;
import java.util.TreeSet;
import java.util.stream.Collectors;
@@ -610,6 +612,23 @@ public class SourceRewriter extends SimplePlanNodeRewriter<DistributionPlanConte
return false;
}
+ @Override
+ public PlanNode visitGroupByTag(GroupByTagNode root, DistributionPlanContext context) {
+ List<SeriesAggregationSourceNode> sources = splitAggregationSourceByPartition(root, context);
+ Map<TRegionReplicaSet, List<SeriesAggregationSourceNode>> sourceGroup =
+ sources.stream().collect(Collectors.groupingBy(SourceNode::getRegionReplicaSet));
+
+ GroupByTagNode newRoot = (GroupByTagNode) root.clone();
+ for (Entry<TRegionReplicaSet, List<SeriesAggregationSourceNode>> entry :
+ sourceGroup.entrySet()) {
+ for (SeriesAggregationSourceNode node : entry.getValue()) {
+ newRoot.addChild(node);
+ }
+ }
+
+ return newRoot;
+ }
+
private GroupByLevelNode groupSourcesForGroupByLevelWithSlidingWindow(
GroupByLevelNode root,
SlidingWindowAggregationNode slidingWindowNode,
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/PlanGraphPrinter.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/PlanGraphPrinter.java
index 508730a854..f96bae755f 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/PlanGraphPrinter.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/PlanGraphPrinter.java
@@ -29,6 +29,8 @@ import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.ExchangeNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.FillNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.FilterNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.GroupByLevelNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.GroupByTagNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.GroupByTagNode.GroupByTagAggregationDescriptor;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.LimitNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.OffsetNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.SlidingWindowAggregationNode;
@@ -48,6 +50,7 @@ import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.Map.Entry;
public class PlanGraphPrinter extends PlanVisitor<List<String>, PlanGraphPrinter.GraphContext> {
@@ -175,6 +178,34 @@ public class PlanGraphPrinter extends PlanVisitor<List<String>, PlanGraphPrinter
return render(node, boxValue, context);
}
+ @Override
+ public List<String> visitGroupByTag(GroupByTagNode node, GraphContext context) {
+ List<String> boxValue = new ArrayList<>();
+ boxValue.add(String.format("GroupByTag-%s", node.getPlanNodeId().getId()));
+ boxValue.add(String.format("Tag keys: %s", node.getTagKeys()));
+ int bucketIdx = 0;
+ for (Entry<List<String>, List<GroupByTagAggregationDescriptor>> entry :
+ node.getTagValuesToAggregationDescriptors().entrySet()) {
+ boxValue.add(String.format("Bucket-%d: %s", bucketIdx, entry.getKey()));
+ int aggregatorIdx = 0;
+ for (GroupByTagAggregationDescriptor descriptor : entry.getValue()) {
+ if (descriptor == null) {
+ boxValue.add(String.format(" Aggregator-%d: NULL", aggregatorIdx));
+ } else {
+ boxValue.add(
+ String.format(
+ " Aggregator-%d: %s, %s",
+ aggregatorIdx, descriptor.getAggregationType(), descriptor.getStep()));
+ boxValue.add(String.format(" Output: %s", descriptor.getOutputColumnNames()));
+ boxValue.add(String.format(" Input: %s", descriptor.getInputExpressions()));
+ }
+ aggregatorIdx += 1;
+ }
+ bucketIdx += 1;
+ }
+ return render(node, boxValue, context);
+ }
+
@Override
public List<String> visitSlidingWindowAggregation(
SlidingWindowAggregationNode node, GraphContext context) {
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/PlanNodeType.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/PlanNodeType.java
index d7f4c3cfb9..9b95eab88b 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/PlanNodeType.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/PlanNodeType.java
@@ -48,6 +48,7 @@ import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.ExchangeNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.FillNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.FilterNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.GroupByLevelNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.GroupByTagNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.LimitNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.OffsetNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.ProjectNode;
@@ -132,7 +133,8 @@ public enum PlanNodeType {
NODE_PATHS_COUNT((short) 51),
INTERNAL_CREATE_TIMESERIES((short) 52),
ACTIVATE_TEMPLATE((short) 53),
- PATHS_USING_TEMPLATE_SCAN((short) 54);
+ PATHS_USING_TEMPLATE_SCAN((short) 54),
+ GROUP_BY_TAG((short) 55);
public static final int BYTES = Short.BYTES;
@@ -291,6 +293,8 @@ public enum PlanNodeType {
return ActivateTemplateNode.deserialize(buffer);
case 54:
return PathsUsingTemplateScanNode.deserialize(buffer);
+ case 55:
+ return GroupByTagNode.deserialize(buffer);
default:
throw new IllegalArgumentException("Invalid node type: " + nodeType);
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/PlanVisitor.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/PlanVisitor.java
index 83fc883018..bda001ace3 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/PlanVisitor.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/PlanVisitor.java
@@ -47,6 +47,7 @@ import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.ExchangeNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.FillNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.FilterNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.GroupByLevelNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.GroupByTagNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.LimitNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.OffsetNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.ProjectNode;
@@ -115,6 +116,10 @@ public abstract class PlanVisitor<R, C> {
return visitPlan(node, context);
}
+ public R visitGroupByTag(GroupByTagNode node, C context) {
+ return visitPlan(node, context);
+ }
+
public R visitSlidingWindowAggregation(SlidingWindowAggregationNode node, C context) {
return visitPlan(node, context);
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/metedata/read/SchemaFetchScanNode.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/metedata/read/SchemaFetchScanNode.java
index 65adccc001..691806b51b 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/metedata/read/SchemaFetchScanNode.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/metedata/read/SchemaFetchScanNode.java
@@ -48,19 +48,21 @@ public class SchemaFetchScanNode extends SourceNode {
private final PartialPath storageGroup;
private final PathPatternTree patternTree;
private final Map<Integer, Template> templateMap;
-
+ private final boolean withTags;
private TRegionReplicaSet schemaRegionReplicaSet;
public SchemaFetchScanNode(
PlanNodeId id,
PartialPath storageGroup,
PathPatternTree patternTree,
- Map<Integer, Template> templateMap) {
+ Map<Integer, Template> templateMap,
+ boolean withTags) {
super(id);
this.storageGroup = storageGroup;
this.patternTree = patternTree;
this.patternTree.constructTree();
this.templateMap = templateMap;
+ this.withTags = withTags;
}
public PartialPath getStorageGroup() {
@@ -85,7 +87,8 @@ public class SchemaFetchScanNode extends SourceNode {
@Override
public PlanNode clone() {
- return new SchemaFetchScanNode(getPlanNodeId(), storageGroup, patternTree, templateMap);
+ return new SchemaFetchScanNode(
+ getPlanNodeId(), storageGroup, patternTree, templateMap, withTags);
}
@Override
@@ -117,6 +120,7 @@ public class SchemaFetchScanNode extends SourceNode {
for (Template template : templateMap.values()) {
template.serialize(byteBuffer);
}
+ ReadWriteIOUtils.write(withTags, byteBuffer);
}
@Override
@@ -128,6 +132,7 @@ public class SchemaFetchScanNode extends SourceNode {
for (Template template : templateMap.values()) {
template.serialize(stream);
}
+ ReadWriteIOUtils.write(withTags, stream);
}
public static SchemaFetchScanNode deserialize(ByteBuffer byteBuffer) {
@@ -142,9 +147,9 @@ public class SchemaFetchScanNode extends SourceNode {
template.deserialize(byteBuffer);
templateMap.put(template.getId(), template);
}
-
+ boolean withTags = ReadWriteIOUtils.readBool(byteBuffer);
PlanNodeId planNodeId = PlanNodeId.deserialize(byteBuffer);
- return new SchemaFetchScanNode(planNodeId, storageGroup, patternTree, templateMap);
+ return new SchemaFetchScanNode(planNodeId, storageGroup, patternTree, templateMap, withTags);
}
@Override
@@ -167,4 +172,8 @@ public class SchemaFetchScanNode extends SourceNode {
public <R, C> R accept(PlanVisitor<R, C> visitor, C context) {
return visitor.visitSchemaFetchScan(this, context);
}
+
+ public boolean isWithTags() {
+ return withTags;
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/process/GroupByTagNode.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/process/GroupByTagNode.java
new file mode 100644
index 0000000000..42144d277e
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/process/GroupByTagNode.java
@@ -0,0 +1,358 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.mpp.plan.planner.plan.node.process;
+
+import org.apache.iotdb.db.mpp.plan.expression.Expression;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNodeId;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNodeType;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanVisitor;
+import org.apache.iotdb.db.mpp.plan.planner.plan.parameter.AggregationDescriptor;
+import org.apache.iotdb.db.mpp.plan.planner.plan.parameter.AggregationStep;
+import org.apache.iotdb.db.mpp.plan.planner.plan.parameter.GroupByTimeParameter;
+import org.apache.iotdb.db.mpp.plan.statement.component.Ordering;
+import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
+
+import org.apache.commons.lang3.Validate;
+
+import javax.annotation.Nullable;
+
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Objects;
+import java.util.stream.Collectors;
+
+public class GroupByTagNode extends MultiChildNode {
+
+ private final List<String> tagKeys;
+ private final Map<List<String>, List<GroupByTagAggregationDescriptor>>
+ tagValuesToAggregationDescriptors;
+ private final List<String> outputColumnNames;
+
+ // The parameter of `group by time`.
+ // Its value will be null if there is no `group by time` clause.
+ @Nullable protected GroupByTimeParameter groupByTimeParameter;
+
+ protected Ordering scanOrder;
+
+ public GroupByTagNode(
+ PlanNodeId id,
+ List<PlanNode> children,
+ @Nullable GroupByTimeParameter groupByTimeParameter,
+ Ordering scanOrder,
+ List<String> tagKeys,
+ Map<List<String>, List<GroupByTagAggregationDescriptor>> tagValuesToAggregationDescriptors,
+ List<String> outputColumnNames) {
+ super(id, children);
+ this.groupByTimeParameter = groupByTimeParameter;
+ this.scanOrder = Validate.notNull(scanOrder);
+ this.tagKeys = Validate.notNull(tagKeys);
+ this.tagValuesToAggregationDescriptors = Validate.notNull(tagValuesToAggregationDescriptors);
+ this.outputColumnNames = Validate.notNull(outputColumnNames);
+ }
+
+ public GroupByTagNode(
+ PlanNodeId id,
+ @Nullable GroupByTimeParameter groupByTimeParameter,
+ Ordering scanOrder,
+ List<String> tagKeys,
+ Map<List<String>, List<GroupByTagAggregationDescriptor>> tagValuesToAggregationDescriptors,
+ List<String> outputColumnNames) {
+ super(id);
+ this.groupByTimeParameter = groupByTimeParameter;
+ this.scanOrder = Validate.notNull(scanOrder);
+ this.tagKeys = Validate.notNull(tagKeys);
+ this.tagValuesToAggregationDescriptors = Validate.notNull(tagValuesToAggregationDescriptors);
+ this.outputColumnNames = Validate.notNull(outputColumnNames);
+ }
+
+ @Override
+ public List<PlanNode> getChildren() {
+ return children;
+ }
+
+ @Override
+ public void addChild(PlanNode child) {
+ this.children.add(child);
+ }
+
+ @Override
+ public PlanNode clone() {
+ // TODO: better do deep copy
+ return new GroupByTagNode(
+ getPlanNodeId(),
+ this.groupByTimeParameter,
+ this.scanOrder,
+ this.tagKeys,
+ this.tagValuesToAggregationDescriptors,
+ this.outputColumnNames);
+ }
+
+ @Override
+ public int allowedChildCount() {
+ return CHILD_COUNT_NO_LIMIT;
+ }
+
+ @Override
+ public List<String> getOutputColumnNames() {
+ List<String> ret = new ArrayList<>(tagKeys);
+ ret.addAll(outputColumnNames);
+ return ret;
+ }
+
+ @Override
+ public <R, C> R accept(PlanVisitor<R, C> visitor, C context) {
+ return visitor.visitGroupByTag(this, context);
+ }
+
+ @Override
+ protected void serializeAttributes(ByteBuffer byteBuffer) {
+ // Plan type.
+ PlanNodeType.GROUP_BY_TAG.serialize(byteBuffer);
+
+ // Tag keys.
+ ReadWriteIOUtils.writeStringList(tagKeys, byteBuffer);
+
+ // Tag values to aggregation descriptors.
+ ReadWriteIOUtils.write(tagValuesToAggregationDescriptors.size(), byteBuffer);
+ for (Entry<List<String>, List<GroupByTagAggregationDescriptor>> entry :
+ tagValuesToAggregationDescriptors.entrySet()) {
+ ReadWriteIOUtils.writeStringList(entry.getKey(), byteBuffer);
+ ReadWriteIOUtils.write(entry.getValue().size(), byteBuffer);
+ for (GroupByTagAggregationDescriptor aggregationDescriptor : entry.getValue()) {
+ if (aggregationDescriptor == null) {
+ ReadWriteIOUtils.write((byte) 0, byteBuffer);
+ } else {
+ ReadWriteIOUtils.write((byte) 1, byteBuffer);
+ aggregationDescriptor.serialize(byteBuffer);
+ }
+ }
+ }
+
+ // Output column names.
+ ReadWriteIOUtils.writeStringList(outputColumnNames, byteBuffer);
+
+ // Group by time parameter.
+ if (groupByTimeParameter == null) {
+ ReadWriteIOUtils.write((byte) 0, byteBuffer);
+ } else {
+ ReadWriteIOUtils.write((byte) 1, byteBuffer);
+ groupByTimeParameter.serialize(byteBuffer);
+ }
+
+ // Scan order.
+ ReadWriteIOUtils.write(scanOrder.ordinal(), byteBuffer);
+ }
+
+ @Override
+ protected void serializeAttributes(DataOutputStream stream) throws IOException {
+ // Plan type.
+ PlanNodeType.GROUP_BY_TAG.serialize(stream);
+
+ // Tag keys.
+ ReadWriteIOUtils.writeStringList(tagKeys, stream);
+
+ // Tag values to aggregation descriptors.
+ ReadWriteIOUtils.write(tagValuesToAggregationDescriptors.size(), stream);
+ for (Entry<List<String>, List<GroupByTagAggregationDescriptor>> entry :
+ tagValuesToAggregationDescriptors.entrySet()) {
+ ReadWriteIOUtils.writeStringList(entry.getKey(), stream);
+ ReadWriteIOUtils.write(entry.getValue().size(), stream);
+ for (GroupByTagAggregationDescriptor aggregationDescriptor : entry.getValue()) {
+ if (aggregationDescriptor == null) {
+ ReadWriteIOUtils.write((byte) 0, stream);
+ } else {
+ ReadWriteIOUtils.write((byte) 1, stream);
+ aggregationDescriptor.serialize(stream);
+ }
+ }
+ }
+
+ // Output column names.
+ ReadWriteIOUtils.writeStringList(outputColumnNames, stream);
+
+ // Group by time parameter.
+ if (groupByTimeParameter == null) {
+ ReadWriteIOUtils.write((byte) 0, stream);
+ } else {
+ ReadWriteIOUtils.write((byte) 1, stream);
+ groupByTimeParameter.serialize(stream);
+ }
+
+ // Scan order.
+ ReadWriteIOUtils.write(scanOrder.ordinal(), stream);
+ }
+
+ @Nullable
+ public GroupByTimeParameter getGroupByTimeParameter() {
+ return groupByTimeParameter;
+ }
+
+ public Ordering getScanOrder() {
+ return scanOrder;
+ }
+
+ public List<String> getTagKeys() {
+ return tagKeys;
+ }
+
+ public Map<List<String>, List<GroupByTagAggregationDescriptor>>
+ getTagValuesToAggregationDescriptors() {
+ return tagValuesToAggregationDescriptors;
+ }
+
+ public static GroupByTagNode deserialize(ByteBuffer byteBuffer) {
+ // Tag keys.
+ List<String> tagKeys = ReadWriteIOUtils.readStringList(byteBuffer);
+
+ // Tag values to aggregation descriptors.
+ int numOfEntries = ReadWriteIOUtils.readInt(byteBuffer);
+ Map<List<String>, List<GroupByTagAggregationDescriptor>> tagValuesToAggregationDescriptors =
+ new HashMap<>();
+ while (numOfEntries > 0) {
+ List<String> tagValues = ReadWriteIOUtils.readStringList(byteBuffer);
+ List<GroupByTagAggregationDescriptor> aggregationDescriptors = new ArrayList<>();
+ int numOfAggregationDescriptors = ReadWriteIOUtils.readInt(byteBuffer);
+ while (numOfAggregationDescriptors > 0) {
+ byte isNotNull = ReadWriteIOUtils.readByte(byteBuffer);
+ if (isNotNull == 1) {
+ aggregationDescriptors.add(GroupByTagAggregationDescriptor.deserialize(byteBuffer));
+ }
+ numOfAggregationDescriptors -= 1;
+ }
+ tagValuesToAggregationDescriptors.put(tagValues, aggregationDescriptors);
+ numOfEntries -= 1;
+ }
+
+ // Output column names.
+ List<String> outputColumnNames = ReadWriteIOUtils.readStringList(byteBuffer);
+
+ // Group by time parameter.
+ byte hasGroupByTimeParameter = ReadWriteIOUtils.readByte(byteBuffer);
+ GroupByTimeParameter groupByTimeParameter = null;
+ if (hasGroupByTimeParameter == 1) {
+ groupByTimeParameter = GroupByTimeParameter.deserialize(byteBuffer);
+ }
+
+ // Scan order.
+ Ordering scanOrder = Ordering.values()[ReadWriteIOUtils.readInt(byteBuffer)];
+
+ PlanNodeId planNodeId = PlanNodeId.deserialize(byteBuffer);
+ return new GroupByTagNode(
+ planNodeId,
+ groupByTimeParameter,
+ scanOrder,
+ tagKeys,
+ tagValuesToAggregationDescriptors,
+ outputColumnNames);
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ if (!super.equals(o)) {
+ return false;
+ }
+ GroupByTagNode that = (GroupByTagNode) o;
+ return Objects.equals(groupByTimeParameter, that.groupByTimeParameter)
+ && scanOrder == that.scanOrder
+ && Objects.equals(tagKeys, that.tagKeys)
+ && Objects.equals(tagValuesToAggregationDescriptors, that.tagValuesToAggregationDescriptors)
+ && Objects.equals(outputColumnNames, that.outputColumnNames);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(
+ super.hashCode(),
+ groupByTimeParameter,
+ scanOrder,
+ tagKeys,
+ tagValuesToAggregationDescriptors,
+ outputColumnNames);
+ }
+
+ @Override
+ public String toString() {
+ return String.format(
+ "GroupByTagNode-%s: Output: %s, Input: %s",
+ getPlanNodeId(),
+ getOutputColumnNames(),
+ tagValuesToAggregationDescriptors.values().stream()
+ .flatMap(list -> list.stream().map(AggregationDescriptor::getInputExpressions))
+ .collect(Collectors.toList()));
+ }
+
+ public static class GroupByTagAggregationDescriptor extends AggregationDescriptor {
+ private final Expression outputExpression;
+
+ public GroupByTagAggregationDescriptor(
+ String aggregationFuncName,
+ AggregationStep step,
+ List<Expression> inputExpressions,
+ Expression outputExpression) {
+ super(aggregationFuncName, step, inputExpressions);
+ this.outputExpression = Validate.notNull(outputExpression);
+ }
+
+ public Expression getOutputExpression() {
+ return outputExpression;
+ }
+
+ @Override
+ public void serialize(ByteBuffer byteBuffer) {
+ super.serialize(byteBuffer);
+ Expression.serialize(outputExpression, byteBuffer);
+ }
+
+ @Override
+ public void serialize(DataOutputStream stream) throws IOException {
+ super.serialize(stream);
+ Expression.serialize(outputExpression, stream);
+ }
+
+ @Override
+ public List<String> getOutputColumnNames() {
+ return Collections.singletonList(outputExpression.toString());
+ }
+
+ public static GroupByTagAggregationDescriptor deserialize(ByteBuffer byteBuffer) {
+ AggregationDescriptor aggregationDescriptor = AggregationDescriptor.deserialize(byteBuffer);
+ Expression outputExpression = Expression.deserialize(byteBuffer);
+ return new GroupByTagAggregationDescriptor(
+ aggregationDescriptor.getAggregationFuncName(),
+ aggregationDescriptor.getStep(),
+ aggregationDescriptor.getInputExpressions(),
+ outputExpression);
+ }
+ }
+}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/parameter/AggregationDescriptor.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/parameter/AggregationDescriptor.java
index daf7955522..16f3896951 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/parameter/AggregationDescriptor.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/parameter/AggregationDescriptor.java
@@ -137,7 +137,7 @@ public class AggregationDescriptor {
outputAggregationNames.add(AggregationType.MAX_TIME.name().toLowerCase());
break;
default:
- outputAggregationNames.add(aggregationFuncName);
+ outputAggregationNames.add(aggregationFuncName.toLowerCase());
}
} else {
outputAggregationNames.add(aggregationFuncName);
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/statement/component/GroupByTagComponent.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/statement/component/GroupByTagComponent.java
new file mode 100644
index 0000000000..32855caec8
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/statement/component/GroupByTagComponent.java
@@ -0,0 +1,38 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.mpp.plan.statement.component;
+
+import org.apache.iotdb.db.mpp.plan.statement.StatementNode;
+
+import org.apache.commons.lang3.Validate;
+
+import java.util.List;
+
+public class GroupByTagComponent extends StatementNode {
+ private final List<String> tagKeys;
+
+ public GroupByTagComponent(List<String> tagKeys) {
+ this.tagKeys = Validate.notNull(tagKeys);
+ }
+
+ public List<String> getTagKeys() {
+ return tagKeys;
+ }
+}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/statement/crud/QueryStatement.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/statement/crud/QueryStatement.java
index 51ad5447e7..d2d84338c1 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/statement/crud/QueryStatement.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/statement/crud/QueryStatement.java
@@ -30,6 +30,7 @@ import org.apache.iotdb.db.mpp.plan.statement.StatementVisitor;
import org.apache.iotdb.db.mpp.plan.statement.component.FillComponent;
import org.apache.iotdb.db.mpp.plan.statement.component.FromComponent;
import org.apache.iotdb.db.mpp.plan.statement.component.GroupByLevelComponent;
+import org.apache.iotdb.db.mpp.plan.statement.component.GroupByTagComponent;
import org.apache.iotdb.db.mpp.plan.statement.component.GroupByTimeComponent;
import org.apache.iotdb.db.mpp.plan.statement.component.HavingCondition;
import org.apache.iotdb.db.mpp.plan.statement.component.OrderByComponent;
@@ -94,6 +95,9 @@ public class QueryStatement extends Statement {
// `GROUP BY LEVEL` clause
protected GroupByLevelComponent groupByLevelComponent;
+ // `GROUP BY TAG` clause
+ protected GroupByTagComponent groupByTagComponent;
+
public QueryStatement() {
this.statementType = StatementType.QUERY;
}
@@ -218,6 +222,14 @@ public class QueryStatement extends Statement {
this.groupByLevelComponent = groupByLevelComponent;
}
+ public GroupByTagComponent getGroupByTagComponent() {
+ return groupByTagComponent;
+ }
+
+ public void setGroupByTagComponent(GroupByTagComponent groupByTagComponent) {
+ this.groupByTagComponent = groupByTagComponent;
+ }
+
public boolean isLastQuery() {
return selectComponent.isHasLast();
}
@@ -230,6 +242,10 @@ public class QueryStatement extends Statement {
return groupByLevelComponent != null;
}
+ public boolean isGroupByTag() {
+ return groupByTagComponent != null;
+ }
+
public boolean isGroupByTime() {
return groupByTimeComponent != null;
}
@@ -276,6 +292,12 @@ public class QueryStatement extends Statement {
if (isGroupByLevel() && isAlignByDevice()) {
throw new SemanticException("group by level does not support align by device now.");
}
+ if (isGroupByTag() && isAlignByDevice()) {
+ throw new SemanticException("group by tag does not support align by device now.");
+ }
+ if (isGroupByTag() && isGroupByLevel()) {
+ throw new SemanticException("group by level cannot be used togather with group by tag");
+ }
for (ResultColumn resultColumn : selectComponent.getResultColumns()) {
if (resultColumn.getColumnType() != ResultColumn.ColumnType.AGGREGATION) {
throw new SemanticException("Raw data and aggregation hybrid query is not supported.");
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/statement/internal/SchemaFetchStatement.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/statement/internal/SchemaFetchStatement.java
index fd55d503c3..dea79e1de7 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/statement/internal/SchemaFetchStatement.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/statement/internal/SchemaFetchStatement.java
@@ -33,11 +33,14 @@ public class SchemaFetchStatement extends Statement {
private final PathPatternTree patternTree;
private final Map<Integer, Template> templateMap;
+ private final boolean withTags;
- public SchemaFetchStatement(PathPatternTree patternTree, Map<Integer, Template> templateMap) {
+ public SchemaFetchStatement(
+ PathPatternTree patternTree, Map<Integer, Template> templateMap, boolean withTags) {
super();
this.patternTree = patternTree;
this.templateMap = templateMap;
+ this.withTags = withTags;
setType(StatementType.FETCH_SCHEMA);
}
@@ -58,4 +61,8 @@ public class SchemaFetchStatement extends Statement {
public List<PartialPath> getPaths() {
return patternTree.getAllPathPatterns();
}
+
+ public boolean isWithTags() {
+ return withTags;
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/strategy/optimizer/ConcatPathOptimizer.java b/server/src/main/java/org/apache/iotdb/db/qp/strategy/optimizer/ConcatPathOptimizer.java
index c25498042d..51c2faccbb 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/strategy/optimizer/ConcatPathOptimizer.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/strategy/optimizer/ConcatPathOptimizer.java
@@ -366,7 +366,8 @@ public class ConcatPathOptimizer implements ILogicalOptimizer {
try {
for (PartialPath originalPath : originalPaths) {
List<MeasurementPath> all =
- IoTDB.schemaProcessor.getMeasurementPathsWithAlias(originalPath, 0, 0, isPrefixMatch)
+ IoTDB.schemaProcessor.getMeasurementPathsWithAlias(
+ originalPath, 0, 0, isPrefixMatch, false)
.left;
if (all.isEmpty()) {
throw new LogicalOptimizeException(
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/utils/WildcardsRemover.java b/server/src/main/java/org/apache/iotdb/db/qp/utils/WildcardsRemover.java
index 30cf3df9f3..5e218b2162 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/utils/WildcardsRemover.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/utils/WildcardsRemover.java
@@ -72,7 +72,7 @@ public class WildcardsRemover {
try {
Pair<List<MeasurementPath>, Integer> pair =
IoTDB.schemaProcessor.getMeasurementPathsWithAlias(
- path, currentLimit, currentOffset, isPrefixMatch);
+ path, currentLimit, currentOffset, isPrefixMatch, false);
consumed += pair.right;
currentOffset -= Math.min(currentOffset, pair.right);
currentLimit -= pair.left.size();
@@ -123,7 +123,11 @@ public class WildcardsRemover {
return remainingExpressions;
}
- /** @return should break the loop or not */
+ /**
+ * Check whether the path number is over limit.
+ *
+ * @return should break the loop or not
+ */
public boolean checkIfPathNumberIsOverLimit(List<ResultColumn> resultColumns)
throws PathNumOverLimitException {
if (resultColumns.size()
diff --git a/server/src/test/java/org/apache/iotdb/db/metadata/cache/DataNodeSchemaCacheTest.java b/server/src/test/java/org/apache/iotdb/db/metadata/cache/DataNodeSchemaCacheTest.java
index 2ed740eb2e..a581cdd867 100644
--- a/server/src/test/java/org/apache/iotdb/db/metadata/cache/DataNodeSchemaCacheTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/metadata/cache/DataNodeSchemaCacheTest.java
@@ -33,15 +33,19 @@ import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
+import java.util.HashMap;
import java.util.Map;
import java.util.stream.Collectors;
public class DataNodeSchemaCacheTest {
DataNodeSchemaCache dataNodeSchemaCache;
+ private Map<String, String> s1TagMap;
@Before
public void setUp() throws Exception {
dataNodeSchemaCache = DataNodeSchemaCache.getInstance();
+ s1TagMap = new HashMap<>();
+ s1TagMap.put("k1", "v1");
}
@After
@@ -67,16 +71,21 @@ public class DataNodeSchemaCacheTest {
o ->
new SchemaCacheEntry(
(MeasurementSchema) o.getMeasurementSchema(),
+ o.getTagMap(),
o.isUnderAlignedEntity())));
Assert.assertEquals(
TSDataType.INT32,
schemaCacheEntryMap.get(new PartialPath("root.sg1.d1.s1")).getTsDataType());
+ Assert.assertEquals(
+ s1TagMap, schemaCacheEntryMap.get(new PartialPath("root.sg1.d1.s1")).getTagMap());
Assert.assertEquals(
TSDataType.FLOAT,
schemaCacheEntryMap.get(new PartialPath("root.sg1.d1.s2")).getTsDataType());
+ Assert.assertNull(schemaCacheEntryMap.get(new PartialPath("root.sg1.d1.s2")).getTagMap());
Assert.assertEquals(
TSDataType.BOOLEAN,
schemaCacheEntryMap.get(new PartialPath("root.sg1.d1.s3")).getTsDataType());
+ Assert.assertNull(schemaCacheEntryMap.get(new PartialPath("root.sg1.d1.s3")).getTagMap());
Assert.assertEquals(3, dataNodeSchemaCache.estimatedSize());
String[] otherMeasurements = new String[3];
@@ -94,16 +103,20 @@ public class DataNodeSchemaCacheTest {
o ->
new SchemaCacheEntry(
(MeasurementSchema) o.getMeasurementSchema(),
+ o.getTagMap(),
o.isUnderAlignedEntity())));
Assert.assertEquals(
TSDataType.BOOLEAN,
schemaCacheEntryMap.get(new PartialPath("root.sg1.d1.s3")).getTsDataType());
+ Assert.assertNull(schemaCacheEntryMap.get(new PartialPath("root.sg1.d1.s3")).getTagMap());
Assert.assertEquals(
TSDataType.TEXT,
schemaCacheEntryMap.get(new PartialPath("root.sg1.d1.s4")).getTsDataType());
+ Assert.assertNull(schemaCacheEntryMap.get(new PartialPath("root.sg1.d1.s4")).getTagMap());
Assert.assertEquals(
TSDataType.INT64,
schemaCacheEntryMap.get(new PartialPath("root.sg1.d1.s5")).getTsDataType());
+ Assert.assertNull(schemaCacheEntryMap.get(new PartialPath("root.sg1.d1.s4")).getTagMap());
Assert.assertEquals(5, dataNodeSchemaCache.estimatedSize());
}
@@ -176,21 +189,25 @@ public class DataNodeSchemaCacheTest {
private ISchemaTree generateSchemaTree1() throws IllegalPathException {
ClusterSchemaTree schemaTree = new ClusterSchemaTree();
-
+ Map<String, String> s1TagMap = new HashMap<>();
+ s1TagMap.put("k1", "v1");
schemaTree.appendSingleMeasurement(
new PartialPath("root.sg1.d1.s1"),
new MeasurementSchema("s1", TSDataType.INT32),
+ s1TagMap,
null,
false);
schemaTree.appendSingleMeasurement(
new PartialPath("root.sg1.d1.s2"),
new MeasurementSchema("s2", TSDataType.FLOAT),
null,
+ null,
false);
schemaTree.appendSingleMeasurement(
new PartialPath("root.sg1.d1.s3"),
new MeasurementSchema("s3", TSDataType.BOOLEAN),
null,
+ null,
false);
return schemaTree;
@@ -203,16 +220,19 @@ public class DataNodeSchemaCacheTest {
new PartialPath("root.sg1.d1.s3"),
new MeasurementSchema("s3", TSDataType.BOOLEAN),
null,
+ null,
false);
schemaTree.appendSingleMeasurement(
new PartialPath("root.sg1.d1.s4"),
new MeasurementSchema("s4", TSDataType.TEXT),
null,
+ null,
false);
schemaTree.appendSingleMeasurement(
new PartialPath("root.sg1.d1.s5"),
new MeasurementSchema("s5", TSDataType.INT64),
null,
+ null,
false);
return schemaTree;
diff --git a/server/src/test/java/org/apache/iotdb/db/metadata/mtree/MTreeBelowSGTest.java b/server/src/test/java/org/apache/iotdb/db/metadata/mtree/MTreeBelowSGTest.java
index a7f4e388ca..dc404680d9 100644
--- a/server/src/test/java/org/apache/iotdb/db/metadata/mtree/MTreeBelowSGTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/metadata/mtree/MTreeBelowSGTest.java
@@ -92,9 +92,11 @@ public abstract class MTreeBelowSGTest {
IMTreeBelowSG mtree;
if (SchemaEngineMode.valueOf(IoTDBDescriptor.getInstance().getConfig().getSchemaEngineMode())
.equals(SchemaEngineMode.Schema_File)) {
- mtree = new MTreeBelowSGCachedImpl(root.getStorageGroupNodeByStorageGroupPath(path), 0);
+ mtree =
+ new MTreeBelowSGCachedImpl(root.getStorageGroupNodeByStorageGroupPath(path), null, 0);
} else {
- mtree = new MTreeBelowSGMemoryImpl(root.getStorageGroupNodeByStorageGroupPath(path), 0);
+ mtree =
+ new MTreeBelowSGMemoryImpl(root.getStorageGroupNodeByStorageGroupPath(path), null, 0);
}
usedMTree.add(mtree);
return mtree;
@@ -283,7 +285,8 @@ public abstract class MTreeBelowSGTest {
assertEquals("root.a.d1.s0", result.get(1).getFullPath());
List<MeasurementPath> result2 =
- storageGroup.getMeasurementPathsWithAlias(new PartialPath("root.a.*.s0"), 0, 0, false)
+ storageGroup.getMeasurementPathsWithAlias(
+ new PartialPath("root.a.*.s0"), 0, 0, false, false)
.left;
result2.sort(Comparator.comparing(MeasurementPath::getFullPath));
assertEquals(2, result2.size());
@@ -294,7 +297,7 @@ public abstract class MTreeBelowSGTest {
result2 =
storageGroup.getMeasurementPathsWithAlias(
- new PartialPath("root.a.*.temperature"), 0, 0, false)
+ new PartialPath("root.a.*.temperature"), 0, 0, false, false)
.left;
result2.sort(Comparator.comparing(MeasurementPath::getFullPath));
assertEquals(2, result2.size());
@@ -302,7 +305,8 @@ public abstract class MTreeBelowSGTest {
assertEquals("root.a.d1.temperature", result2.get(1).getFullPathWithAlias());
Pair<List<MeasurementPath>, Integer> result3 =
- storageGroup.getMeasurementPathsWithAlias(new PartialPath("root.a.**"), 2, 0, false);
+ storageGroup.getMeasurementPathsWithAlias(
+ new PartialPath("root.a.**"), 2, 0, false, false);
assertEquals(2, result3.left.size());
assertEquals(2, result3.right.intValue());
@@ -516,7 +520,7 @@ public abstract class MTreeBelowSGTest {
assertEquals(
2,
storageGroup
- .getMeasurementPathsWithAlias(new PartialPath("root.**"), 0, 0, false)
+ .getMeasurementPathsWithAlias(new PartialPath("root.**"), 0, 0, false, false)
.left
.size());
}
diff --git a/server/src/test/java/org/apache/iotdb/db/metadata/path/MeasurementPathTest.java b/server/src/test/java/org/apache/iotdb/db/metadata/path/MeasurementPathTest.java
new file mode 100644
index 0000000000..475950325c
--- /dev/null
+++ b/server/src/test/java/org/apache/iotdb/db/metadata/path/MeasurementPathTest.java
@@ -0,0 +1,104 @@
+/*
+ * 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.metadata.path;
+
+import org.apache.iotdb.commons.exception.IllegalPathException;
+import org.apache.iotdb.commons.path.PartialPath;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
+import org.apache.iotdb.tsfile.write.schema.IMeasurementSchema;
+import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.io.ByteArrayOutputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.HashMap;
+
+public class MeasurementPathTest {
+
+ @Test
+ public void testMeasurementPathSerde() throws IllegalPathException, IOException {
+ IMeasurementSchema schema = new MeasurementSchema("s1", TSDataType.TEXT);
+ MeasurementPath expectedPath =
+ new MeasurementPath(new PartialPath("root.sg.d1.s1"), schema, true);
+ expectedPath.setMeasurementAlias("alias_s1");
+ MeasurementPath actualPath = serdeWithByteBuffer(expectedPath);
+ assertMeasurementPathEquals(actualPath, expectedPath);
+ actualPath = serdeWithStream(expectedPath);
+ assertMeasurementPathEquals(actualPath, expectedPath);
+
+ HashMap<String, String> tagMap = new HashMap<>();
+ tagMap.put("k1", "v1");
+ expectedPath.setTagMap(tagMap);
+ actualPath = serdeWithByteBuffer(expectedPath);
+ assertMeasurementPathEquals(actualPath, expectedPath);
+ actualPath = serdeWithStream(expectedPath);
+ assertMeasurementPathEquals(actualPath, expectedPath);
+ }
+
+ @Test
+ public void testCloneAndCopy() throws IllegalPathException {
+ IMeasurementSchema schema = new MeasurementSchema("s1", TSDataType.TEXT);
+ MeasurementPath expectedPath =
+ new MeasurementPath(new PartialPath("root.sg.d1.s1"), schema, true);
+ expectedPath.setMeasurementAlias("alias_s1");
+
+ MeasurementPath actualPath = expectedPath.clone();
+ assertMeasurementPathEquals(actualPath, expectedPath);
+ actualPath = (MeasurementPath) expectedPath.copy();
+ assertMeasurementPathEquals(actualPath, expectedPath);
+
+ HashMap<String, String> tagMap = new HashMap<>();
+ tagMap.put("k1", "v1");
+ expectedPath.setTagMap(tagMap);
+ actualPath = expectedPath.clone();
+ assertMeasurementPathEquals(actualPath, expectedPath);
+ actualPath = (MeasurementPath) expectedPath.copy();
+ assertMeasurementPathEquals(actualPath, expectedPath);
+ }
+
+ private MeasurementPath serdeWithByteBuffer(MeasurementPath origin) {
+ ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
+ origin.serialize(byteBuffer);
+ byteBuffer.flip();
+ Assert.assertEquals((byte) 0, ReadWriteIOUtils.readByte(byteBuffer));
+ return MeasurementPath.deserialize(byteBuffer);
+ }
+
+ private MeasurementPath serdeWithStream(MeasurementPath origin) throws IOException {
+ ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
+ DataOutputStream dataOutputStream = new DataOutputStream(byteArrayOutputStream);
+ origin.serialize(dataOutputStream);
+ ByteBuffer byteBuffer = ByteBuffer.wrap(byteArrayOutputStream.toByteArray());
+ Assert.assertEquals((byte) 0, ReadWriteIOUtils.readByte(byteBuffer));
+ return MeasurementPath.deserialize(byteBuffer);
+ }
+
+ private void assertMeasurementPathEquals(MeasurementPath actual, MeasurementPath expected) {
+ Assert.assertEquals(expected, actual);
+ Assert.assertEquals(expected.isUnderAlignedEntity(), actual.isUnderAlignedEntity());
+ Assert.assertEquals(expected.getTagMap(), actual.getTagMap());
+ Assert.assertEquals(expected.getMeasurementSchema(), actual.getMeasurementSchema());
+ Assert.assertEquals(expected.getMeasurementAlias(), actual.getMeasurementAlias());
+ }
+}
diff --git a/server/src/test/java/org/apache/iotdb/db/mpp/execution/operator/OperatorMemoryTest.java b/server/src/test/java/org/apache/iotdb/db/mpp/execution/operator/OperatorMemoryTest.java
index ffb8eae2e5..1f04825001 100644
--- a/server/src/test/java/org/apache/iotdb/db/mpp/execution/operator/OperatorMemoryTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/mpp/execution/operator/OperatorMemoryTest.java
@@ -932,7 +932,12 @@ public class OperatorMemoryTest {
SchemaFetchScanOperator operator =
new SchemaFetchScanOperator(
- planNodeId, fragmentInstanceContext.getOperatorContexts().get(0), null, null, null);
+ planNodeId,
+ fragmentInstanceContext.getOperatorContexts().get(0),
+ null,
+ null,
+ null,
+ false);
assertEquals(DEFAULT_MAX_TSBLOCK_SIZE_IN_BYTES, operator.calculateMaxPeekMemory());
assertEquals(DEFAULT_MAX_TSBLOCK_SIZE_IN_BYTES, operator.calculateMaxReturnSize());
diff --git a/server/src/test/java/org/apache/iotdb/db/mpp/execution/operator/schema/SchemaFetchScanOperatorTest.java b/server/src/test/java/org/apache/iotdb/db/mpp/execution/operator/schema/SchemaFetchScanOperatorTest.java
index a99a4743b9..72cff46add 100644
--- a/server/src/test/java/org/apache/iotdb/db/mpp/execution/operator/schema/SchemaFetchScanOperatorTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/mpp/execution/operator/schema/SchemaFetchScanOperatorTest.java
@@ -74,7 +74,8 @@ public class SchemaFetchScanOperatorTest {
patternTree.constructTree();
SchemaFetchScanOperator schemaFetchScanOperator =
- new SchemaFetchScanOperator(null, null, patternTree, Collections.emptyMap(), schemaRegion);
+ new SchemaFetchScanOperator(
+ null, null, patternTree, Collections.emptyMap(), schemaRegion, false);
Assert.assertTrue(schemaFetchScanOperator.hasNext());
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/FakePartitionFetcherImpl.java b/server/src/test/java/org/apache/iotdb/db/mpp/plan/analyze/FakePartitionFetcherImpl.java
similarity index 100%
rename from server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/FakePartitionFetcherImpl.java
rename to server/src/test/java/org/apache/iotdb/db/mpp/plan/analyze/FakePartitionFetcherImpl.java
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/FakeSchemaFetcherImpl.java b/server/src/test/java/org/apache/iotdb/db/mpp/plan/analyze/FakeSchemaFetcherImpl.java
similarity index 92%
rename from server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/FakeSchemaFetcherImpl.java
rename to server/src/test/java/org/apache/iotdb/db/mpp/plan/analyze/FakeSchemaFetcherImpl.java
index 7c620a43e3..3b915ad745 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/FakeSchemaFetcherImpl.java
+++ b/server/src/test/java/org/apache/iotdb/db/mpp/plan/analyze/FakeSchemaFetcherImpl.java
@@ -47,6 +47,11 @@ public class FakeSchemaFetcherImpl implements ISchemaFetcher {
return schemaTree;
}
+ @Override
+ public ISchemaTree fetchSchemaWithTags(PathPatternTree patternTree) {
+ return fetchSchema(patternTree);
+ }
+
@Override
public ISchemaTree fetchSchemaWithAutoCreate(
PartialPath devicePath,
@@ -70,12 +75,16 @@ public class FakeSchemaFetcherImpl implements ISchemaFetcher {
SchemaMeasurementNode s1 =
new SchemaMeasurementNode("s1", new MeasurementSchema("s1", TSDataType.INT32));
+ s1.setTagMap(Collections.singletonMap("key1", "value1"));
SchemaMeasurementNode s2 =
new SchemaMeasurementNode("s2", new MeasurementSchema("s2", TSDataType.DOUBLE));
+ s2.setTagMap(Collections.singletonMap("key1", "value1"));
SchemaMeasurementNode s3 =
new SchemaMeasurementNode("s3", new MeasurementSchema("s3", TSDataType.BOOLEAN));
+ s3.setTagMap(Collections.singletonMap("key1", "value2"));
SchemaMeasurementNode s4 =
new SchemaMeasurementNode("s4", new MeasurementSchema("s4", TSDataType.TEXT));
+ s4.setTagMap(Collections.singletonMap("key2", "value1"));
s2.setAlias("status");
SchemaEntityNode d1 = new SchemaEntityNode("d1");
diff --git a/server/src/test/java/org/apache/iotdb/db/mpp/plan/parser/StatementGeneratorTest.java b/server/src/test/java/org/apache/iotdb/db/mpp/plan/parser/StatementGeneratorTest.java
index 32e992ef1d..eecb997038 100644
--- a/server/src/test/java/org/apache/iotdb/db/mpp/plan/parser/StatementGeneratorTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/mpp/plan/parser/StatementGeneratorTest.java
@@ -20,6 +20,7 @@
package org.apache.iotdb.db.mpp.plan.parser;
import org.apache.iotdb.commons.path.PartialPath;
+import org.apache.iotdb.db.exception.sql.SemanticException;
import org.apache.iotdb.db.mpp.plan.statement.component.ResultColumn;
import org.apache.iotdb.db.mpp.plan.statement.crud.QueryStatement;
@@ -41,6 +42,21 @@ public class StatementGeneratorTest {
"SELECT s1, s2 FROM root.sg1.d1 LIMIT 10 OFFSET 10", selectExprList, prefixPaths, 10, 10);
}
+ @Test
+ public void testGroupByTagWithDuplicatedKeys() {
+ try {
+ checkQueryStatement(
+ "SELECT avg(*) FROM root.sg.** GROUP BY TAGS k1, k2, k1",
+ Collections.emptyList(),
+ Collections.emptyList(),
+ 10,
+ 10);
+ Assert.fail();
+ } catch (SemanticException e) {
+ Assert.assertEquals("duplicated key in GROUP BY TAGS: k1", e.getMessage());
+ }
+ }
+
// TODO: add more tests
private void checkQueryStatement(
diff --git a/server/src/test/java/org/apache/iotdb/db/mpp/plan/plan/LogicalPlannerTest.java b/server/src/test/java/org/apache/iotdb/db/mpp/plan/plan/LogicalPlannerTest.java
index 1646afe43f..179421b5a7 100644
--- a/server/src/test/java/org/apache/iotdb/db/mpp/plan/plan/LogicalPlannerTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/mpp/plan/plan/LogicalPlannerTest.java
@@ -27,6 +27,8 @@ import org.apache.iotdb.db.mpp.plan.analyze.Analysis;
import org.apache.iotdb.db.mpp.plan.analyze.Analyzer;
import org.apache.iotdb.db.mpp.plan.analyze.FakePartitionFetcherImpl;
import org.apache.iotdb.db.mpp.plan.analyze.FakeSchemaFetcherImpl;
+import org.apache.iotdb.db.mpp.plan.expression.Expression;
+import org.apache.iotdb.db.mpp.plan.expression.leaf.TimeSeriesOperand;
import org.apache.iotdb.db.mpp.plan.parser.StatementGenerator;
import org.apache.iotdb.db.mpp.plan.plan.node.PlanNodeDeserializeHelper;
import org.apache.iotdb.db.mpp.plan.planner.LogicalPlanner;
@@ -44,11 +46,18 @@ import org.apache.iotdb.db.mpp.plan.planner.plan.node.metedata.write.AlterTimeSe
import org.apache.iotdb.db.mpp.plan.planner.plan.node.metedata.write.CreateAlignedTimeSeriesNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.metedata.write.CreateMultiTimeSeriesNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.metedata.write.CreateTimeSeriesNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.GroupByTagNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.GroupByTagNode.GroupByTagAggregationDescriptor;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.LimitNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.OffsetNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.source.AlignedSeriesAggregationScanNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.source.SeriesAggregationScanNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.parameter.AggregationStep;
import org.apache.iotdb.db.mpp.plan.statement.Statement;
+import org.apache.iotdb.db.mpp.plan.statement.component.Ordering;
import org.apache.iotdb.db.mpp.plan.statement.metadata.AlterTimeSeriesStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.CreateMultiTimeSeriesStatement;
+import org.apache.iotdb.db.query.aggregation.AggregationType;
import org.apache.iotdb.service.rpc.thrift.TSCreateMultiTimeseriesReq;
import org.apache.iotdb.tsfile.file.metadata.enums.CompressionType;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
@@ -60,7 +69,10 @@ import org.junit.Test;
import java.nio.ByteBuffer;
import java.time.ZonedDateTime;
import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
import java.util.HashMap;
+import java.util.List;
import java.util.Map;
import static org.apache.iotdb.db.mpp.plan.plan.QueryLogicalPlanUtil.querySQLs;
@@ -73,8 +85,13 @@ public class LogicalPlannerTest {
@Test
public void testQueryPlan() {
for (String sql : querySQLs) {
- Assert.assertEquals(sqlToPlanMap.get(sql), parseSQLToPlanNode(sql));
- System.out.printf("\"%s\" TEST PASSED\n", sql);
+ try {
+ Assert.assertEquals(sqlToPlanMap.get(sql), parseSQLToPlanNode(sql));
+ } catch (Exception e) {
+ System.err.println("Failed to generated logical plan for " + sql);
+ e.printStackTrace();
+ break;
+ }
}
}
@@ -608,21 +625,71 @@ public class LogicalPlannerTest {
}
}
- private PlanNode parseSQLToPlanNode(String sql) {
- PlanNode planNode = null;
+ @Test
+ public void testGroupByTag() {
+ String sql = "select max_value(s1) from root.** group by tags key1";
try {
- Statement statement =
- StatementGenerator.createStatement(sql, ZonedDateTime.now().getOffset());
- MPPQueryContext context = new MPPQueryContext(new QueryId("test_query"));
- Analyzer analyzer =
- new Analyzer(context, new FakePartitionFetcherImpl(), new FakeSchemaFetcherImpl());
- Analysis analysis = analyzer.analyze(statement);
- LogicalPlanner planner = new LogicalPlanner(context, new ArrayList<>());
- planNode = planner.plan(analysis).getRootNode();
+ PlanNode pn = parseSQLToPlanNode(sql);
+ GroupByTagNode root = (GroupByTagNode) pn;
+
+ Assert.assertEquals(Collections.singletonList("key1"), root.getTagKeys());
+
+ Map<List<String>, List<GroupByTagAggregationDescriptor>> tagValuesToAggregationDescriptors =
+ root.getTagValuesToAggregationDescriptors();
+ Assert.assertEquals(1, tagValuesToAggregationDescriptors.size());
+ Assert.assertEquals(
+ Collections.singleton(Collections.singletonList("value1")),
+ tagValuesToAggregationDescriptors.keySet());
+ List<GroupByTagAggregationDescriptor> descriptors =
+ tagValuesToAggregationDescriptors.get(Collections.singletonList("value1"));
+ Assert.assertEquals(1, descriptors.size());
+ GroupByTagAggregationDescriptor descriptor = descriptors.get(0);
+ Assert.assertEquals("max_value(s1)", descriptor.getOutputExpression().toString());
+ Assert.assertEquals(AggregationType.MAX_VALUE, descriptor.getAggregationType());
+ Assert.assertEquals(AggregationStep.FINAL, descriptor.getStep());
+ Assert.assertEquals(3, descriptor.getInputExpressions().size());
+ for (Expression expression : descriptor.getInputExpressions()) {
+ Assert.assertTrue(expression instanceof TimeSeriesOperand);
+ Assert.assertEquals("s1", ((TimeSeriesOperand) expression).getPath().getMeasurement());
+ }
+
+ Assert.assertEquals(Arrays.asList("key1", "max_value(s1)"), root.getOutputColumnNames());
+
+ Assert.assertNull(root.getGroupByTimeParameter());
+
+ Assert.assertEquals(Ordering.ASC, root.getScanOrder());
+
+ Assert.assertEquals(3, root.getChildren().size());
+ for (PlanNode child : root.getChildren()) {
+ Assert.assertTrue(
+ child instanceof AlignedSeriesAggregationScanNode
+ || child instanceof SeriesAggregationScanNode);
+ }
} catch (Exception e) {
e.printStackTrace();
fail();
}
- return planNode;
+ }
+
+ @Test
+ public void testGroupByTagWithValueFilter() {
+ String sql = "select max_value(s1) from root.** where s1>1 group by tags key1";
+ try {
+ parseSQLToPlanNode(sql);
+ fail();
+ } catch (Exception e) {
+ Assert.assertTrue(
+ e.getMessage().contains("Only time filters are supported in GROUP BY TAGS query"));
+ }
+ }
+
+ private PlanNode parseSQLToPlanNode(String sql) {
+ Statement statement = StatementGenerator.createStatement(sql, ZonedDateTime.now().getOffset());
+ MPPQueryContext context = new MPPQueryContext(new QueryId("test_query"));
+ Analyzer analyzer =
+ new Analyzer(context, new FakePartitionFetcherImpl(), new FakeSchemaFetcherImpl());
+ Analysis analysis = analyzer.analyze(statement);
+ LogicalPlanner planner = new LogicalPlanner(context, new ArrayList<>());
+ return planner.plan(analysis).getRootNode();
}
}
diff --git a/server/src/test/java/org/apache/iotdb/db/mpp/plan/plan/node/metadata/read/SchemaFetchScanNodeTest.java b/server/src/test/java/org/apache/iotdb/db/mpp/plan/plan/node/metadata/read/SchemaFetchScanNodeTest.java
index fdc4806e6d..646bc10cd8 100644
--- a/server/src/test/java/org/apache/iotdb/db/mpp/plan/plan/node/metadata/read/SchemaFetchScanNodeTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/mpp/plan/plan/node/metadata/read/SchemaFetchScanNodeTest.java
@@ -40,7 +40,11 @@ public class SchemaFetchScanNodeTest {
patternTree.appendPathPattern(new PartialPath("root.sg.**.*"));
SchemaFetchScanNode schemaFetchScanNode =
new SchemaFetchScanNode(
- new PlanNodeId("0"), new PartialPath("root.sg"), patternTree, Collections.emptyMap());
+ new PlanNodeId("0"),
+ new PartialPath("root.sg"),
+ patternTree,
+ Collections.emptyMap(),
+ true);
ByteBuffer byteBuffer = ByteBuffer.allocate(1024 * 1024);
schemaFetchScanNode.serialize(byteBuffer);
byteBuffer.flip();
@@ -48,5 +52,6 @@ public class SchemaFetchScanNodeTest {
Assert.assertEquals("root.sg", recoveredNode.getStorageGroup().getFullPath());
Assert.assertEquals(
"root.sg.**.*", recoveredNode.getPatternTree().getAllPathPatterns().get(0).getFullPath());
+ Assert.assertTrue(recoveredNode.isWithTags());
}
}
diff --git a/server/src/test/java/org/apache/iotdb/db/mpp/plan/plan/node/process/GroupByTagNodeSerdeTest.java b/server/src/test/java/org/apache/iotdb/db/mpp/plan/plan/node/process/GroupByTagNodeSerdeTest.java
new file mode 100644
index 0000000000..84edace4c5
--- /dev/null
+++ b/server/src/test/java/org/apache/iotdb/db/mpp/plan/plan/node/process/GroupByTagNodeSerdeTest.java
@@ -0,0 +1,121 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iotdb.db.mpp.plan.plan.node.process;
+
+import org.apache.iotdb.commons.exception.IllegalPathException;
+import org.apache.iotdb.commons.path.PartialPath;
+import org.apache.iotdb.db.metadata.path.MeasurementPath;
+import org.apache.iotdb.db.mpp.plan.expression.leaf.TimeSeriesOperand;
+import org.apache.iotdb.db.mpp.plan.expression.multi.FunctionExpression;
+import org.apache.iotdb.db.mpp.plan.plan.node.PlanNodeDeserializeHelper;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNodeId;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.GroupByTagNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.source.SeriesAggregationScanNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.parameter.AggregationDescriptor;
+import org.apache.iotdb.db.mpp.plan.planner.plan.parameter.AggregationStep;
+import org.apache.iotdb.db.mpp.plan.planner.plan.parameter.GroupByTimeParameter;
+import org.apache.iotdb.db.mpp.plan.statement.component.Ordering;
+import org.apache.iotdb.db.query.aggregation.AggregationType;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.io.ByteArrayOutputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+
+public class GroupByTagNodeSerdeTest {
+
+ @Test
+ public void testSerializeAndDeserialize() throws IllegalPathException, IOException {
+ GroupByTimeParameter groupByTimeParameter =
+ new GroupByTimeParameter(1, 100, 1, 1, true, true, true);
+ GroupByTagNode.GroupByTagAggregationDescriptor s1MaxTime =
+ new GroupByTagNode.GroupByTagAggregationDescriptor(
+ AggregationType.MAX_TIME.name().toLowerCase(),
+ AggregationStep.FINAL,
+ Collections.singletonList(new TimeSeriesOperand(new PartialPath("root.sg.d1.s1"))),
+ new FunctionExpression(
+ "max_time",
+ new LinkedHashMap<>(),
+ Collections.singletonList(
+ new TimeSeriesOperand(new PartialPath("root.sg.d1.s1")))));
+ GroupByTagNode.GroupByTagAggregationDescriptor s1Avg =
+ new GroupByTagNode.GroupByTagAggregationDescriptor(
+ AggregationType.AVG.name().toLowerCase(),
+ AggregationStep.FINAL,
+ Collections.singletonList(new TimeSeriesOperand(new PartialPath("root.sg.d1.s1"))),
+ new FunctionExpression(
+ "avg",
+ new LinkedHashMap<>(),
+ Collections.singletonList(
+ new TimeSeriesOperand(new PartialPath("root.sg.d1.s1")))));
+ AggregationDescriptor s1MaxTimePartial =
+ new AggregationDescriptor(
+ AggregationType.MAX_TIME.name().toLowerCase(),
+ AggregationStep.PARTIAL,
+ Collections.singletonList(new TimeSeriesOperand(new PartialPath("root.sg.d1.s1"))));
+ AggregationDescriptor s1AvgTimePartial =
+ new AggregationDescriptor(
+ AggregationType.AVG.name().toLowerCase(),
+ AggregationStep.PARTIAL,
+ Collections.singletonList(new TimeSeriesOperand(new PartialPath("root.sg.d1.s1"))));
+ Map<List<String>, List<GroupByTagNode.GroupByTagAggregationDescriptor>>
+ tagValuesToAggregationDescriptors = new HashMap<>();
+ tagValuesToAggregationDescriptors.put(
+ Arrays.asList("v1", "v2"), Arrays.asList(s1MaxTime, s1Avg));
+ GroupByTagNode expectedNode =
+ new GroupByTagNode(
+ new PlanNodeId("TestGroupByTagNode"),
+ Collections.singletonList(
+ new SeriesAggregationScanNode(
+ new PlanNodeId("TestSeriesAggregateScanNode1"),
+ new MeasurementPath("root.sg.d1.s1", TSDataType.INT32),
+ Arrays.asList(s1MaxTimePartial, s1AvgTimePartial),
+ Ordering.ASC,
+ null,
+ groupByTimeParameter,
+ null)),
+ groupByTimeParameter,
+ Ordering.ASC,
+ Collections.singletonList("k1"),
+ tagValuesToAggregationDescriptors,
+ Arrays.asList("MAX_TIME(root.sg.d1.s1)", "AVG(root.sg.d1.s1)"));
+
+ ByteBuffer byteBuffer = ByteBuffer.allocate(4096);
+ expectedNode.serialize(byteBuffer);
+ byteBuffer.flip();
+ Assert.assertEquals(expectedNode, PlanNodeDeserializeHelper.deserialize(byteBuffer));
+
+ ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ DataOutputStream dataOutputStream = new DataOutputStream(baos);
+ expectedNode.serialize(dataOutputStream);
+ byte[] byteArray = baos.toByteArray();
+ ByteBuffer buffer = ByteBuffer.wrap(byteArray);
+ Assert.assertEquals(expectedNode, PlanNodeDeserializeHelper.deserialize(buffer));
+ }
+}
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/utils/ReadWriteIOUtils.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/utils/ReadWriteIOUtils.java
index 1ad3b5fbec..45c514d2fa 100644
--- a/tsfile/src/main/java/org/apache/iotdb/tsfile/utils/ReadWriteIOUtils.java
+++ b/tsfile/src/main/java/org/apache/iotdb/tsfile/utils/ReadWriteIOUtils.java
@@ -676,6 +676,22 @@ public class ReadWriteIOUtils {
return bytes;
}
+ public static Map<String, String> readMap(InputStream inputStream) throws IOException {
+ int length = readInt(inputStream);
+ if (length == NO_BYTE_TO_READ) {
+ return null;
+ }
+ Map<String, String> map = new HashMap<>(length);
+ for (int i = 0; i < length; i++) {
+ // key
+ String key = readString(inputStream);
+ // value
+ String value = readString(inputStream);
+ map.put(key, value);
+ }
+ return map;
+ }
+
public static Map<String, String> readMap(ByteBuffer buffer) {
int length = readInt(buffer);
if (length == NO_BYTE_TO_READ) {
@@ -828,7 +844,7 @@ public class ReadWriteIOUtils {
int size = list.size();
buffer.putInt(size);
for (String s : list) {
- buffer.put(s.getBytes());
+ write(s, buffer);
}
}