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/10/14 07:49:56 UTC
[iotdb] branch feature/iotdb-4639 created (now 70e268f105)
This is an automated email from the ASF dual-hosted git repository.
ericpai pushed a change to branch feature/iotdb-4639
in repository https://gitbox.apache.org/repos/asf/iotdb.git
at 70e268f105 [IOTDB-4639] Support limit, slimit, offset and soffset in tag aggregation
This branch includes the following new commits:
new 70e268f105 [IOTDB-4639] Support limit, slimit, offset and soffset in tag aggregation
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
[iotdb] 01/01: [IOTDB-4639] Support limit, slimit, offset and soffset in tag aggregation
Posted by er...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ericpai pushed a commit to branch feature/iotdb-4639
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 70e268f105d302948f62f7f2231ccff941b64481
Author: ericpai <er...@hotmail.com>
AuthorDate: Fri Oct 14 15:49:29 2022 +0800
[IOTDB-4639] Support limit, slimit, offset and soffset in tag aggregation
---
docs/UserGuide/Query-Data/Aggregate-Query.md | 62 +++--
docs/zh/UserGuide/Query-Data/Aggregate-Query.md | 61 +++--
.../db/it/aggregation/IoTDBTagAggregationIT.java | 264 +++++++++------------
.../operator/process/TagAggregationOperator.java | 2 +-
.../db/mpp/plan/planner/LogicalPlanBuilder.java | 6 +-
.../planner/plan/node/process/GroupByTagNode.java | 59 ++++-
.../plan/node/process/GroupByTagNodeSerdeTest.java | 39 ++-
7 files changed, 303 insertions(+), 190 deletions(-)
diff --git a/docs/UserGuide/Query-Data/Aggregate-Query.md b/docs/UserGuide/Query-Data/Aggregate-Query.md
index b23d21ef38..7b04b13a97 100644
--- a/docs/UserGuide/Query-Data/Aggregate-Query.md
+++ b/docs/UserGuide/Query-Data/Aggregate-Query.md
@@ -540,7 +540,7 @@ The results are
+--------+------------------+
| Beijing|104.04666697184244|
|Shanghai|107.85000076293946|
-| NULL| 50.84999910990397|
+| null| 50.84999910990397|
+--------+------------------+
Total line number = 3
It costs 0.231s
@@ -549,7 +549,7 @@ 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.
+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
@@ -571,11 +571,11 @@ The results
+--------+--------+------------------+
| city|workshop| avg(temperature)|
+--------+--------+------------------+
-| NULL| NULL| 50.84999910990397|
-|Shanghai| w1|113.01666768391927|
+| Beijing| w1|103.73750019073486|
| Beijing| w2| 104.4000004359654|
+|Shanghai| w1|113.01666768391927|
|Shanghai| w2|100.10000038146973|
-| Beijing| w1|103.73750019073486|
+| null| null| 50.84999910990397|
+--------+--------+------------------+
Total line number = 5
It costs 0.027s
@@ -601,16 +601,51 @@ 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| w1|103.81666692097981|
|1970-01-01T08:00:01.000+08:00| Beijing| w2| 103.4|
+|1970-01-01T08:00:01.000+08:00|Shanghai| w1|113.20000076293945|
|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:01.000+08:00| null| null| 50.91999893188476|
+|1970-01-01T08:00:06.000+08:00| Beijing| w1| 103.5|
|1970-01-01T08:00:06.000+08:00| Beijing| w2| 106.9000015258789|
+|1970-01-01T08:00:06.000+08:00|Shanghai| w1| 112.6500015258789|
|1970-01-01T08:00:06.000+08:00|Shanghai| w2| 99.80000305175781|
+|1970-01-01T08:00:06.000+08:00| null| null| 50.5|
++-----------------------------+--------+--------+------------------+
+```
+
+### The Order of the Output of Aggregation Query by Tags
+
+Under one aggregate time window, the tag values are compared one by one according to the keys' order specified in `GROUP BY TAGS`.
+The order is the one of value string comparison. The string comparison is implemented by [`String::compareTo`](https://docs.oracle.com/javase/8/docs/api/java/lang/String.html#compareTo-java.lang.String-) in Java.
+
+When executing an aggregation query based on time window, the `ORDER BY TIME` clause can be added to indicate the time output order.
+Different time windows will be output in the specified order, and the records in each window will be output in the tag comparison order.
+
+The above query can be output in a time descending order.
+
+SQL
+
+```SQL
+SELECT AVG(temperature) FROM root.factory1.** GROUP BY ([1000, 10000), 5s), TAGS(city, workshop) ORDER BY TIME DESC;
+```
+
+The results
+
+```
++-----------------------------+--------+--------+------------------+
+| Time| city|workshop| avg(temperature)|
++-----------------------------+--------+--------+------------------+
|1970-01-01T08:00:06.000+08:00| Beijing| w1| 103.5|
+|1970-01-01T08:00:06.000+08:00| Beijing| w2| 106.9000015258789|
+|1970-01-01T08:00:06.000+08:00|Shanghai| w1| 112.6500015258789|
+|1970-01-01T08:00:06.000+08:00|Shanghai| w2| 99.80000305175781|
+|1970-01-01T08:00:06.000+08:00| null| null| 50.5|
+|1970-01-01T08:00:01.000+08:00| Beijing| w1|103.81666692097981|
+|1970-01-01T08:00:01.000+08:00| Beijing| w2| 103.4|
+|1970-01-01T08:00:01.000+08:00|Shanghai| w1|113.20000076293945|
+|1970-01-01T08:00:01.000+08:00|Shanghai| w2| 100.1999994913737|
+|1970-01-01T08:00:01.000+08:00| null| null| 50.91999893188476|
+-----------------------------+--------+--------+------------------+
```
@@ -623,10 +658,9 @@ As this feature is still under development, some queries have not been completed
> 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.
+> 3. Temporarily not support `ALIGN BY DEVICE`.
+> 4. Temporarily not support expressions as aggregation function parameter，e.g. `count(s+1)`.
+> 5. Not support the value filter, which stands the same with the `GROUP BY LEVEL` query.
## Aggregate result filtering
diff --git a/docs/zh/UserGuide/Query-Data/Aggregate-Query.md b/docs/zh/UserGuide/Query-Data/Aggregate-Query.md
index db44805857..269cd601fc 100644
--- a/docs/zh/UserGuide/Query-Data/Aggregate-Query.md
+++ b/docs/zh/UserGuide/Query-Data/Aggregate-Query.md
@@ -529,7 +529,7 @@ SELECT AVG(temperature) FROM root.factory1.** GROUP BY TAGS(city);
+--------+------------------+
| Beijing|104.04666697184244|
|Shanghai|107.85000076293946|
-| NULL| 50.84999910990397|
+| null| 50.84999910990397|
+--------+------------------+
Total line number = 3
It costs 0.231s
@@ -538,7 +538,7 @@ It costs 0.231s
从结果集中可以看到，和时间区间聚合、按层次聚合相比，标签聚合的查询结果的不同点是：
1. 标签聚合查询的聚合结果不会再做去星号展开，而是将多个时间序列的数据作为一个整体进行聚合计算。
2. 标签聚合查询除了输出聚合结果列，还会输出聚合标签的键值列。该列的列名为聚合指定的标签键，列的值则为所有查询的时间序列中出现的该标签的值。
-如果某些时间序列未设置该标签，则在键值列中有一行单独的 `NULL` ，代表未设置标签的所有时间序列数据的聚合结果。
+如果某些时间序列未设置该标签，则在键值列中有一行单独空值 ，代表未设置标签的所有时间序列数据的聚合结果。
### 多标签聚合查询
@@ -558,11 +558,11 @@ SELECT avg(temperature) FROM root.factory1.** GROUP BY TAGS(city, workshop);
+--------+--------+------------------+
| city|workshop| avg(temperature)|
+--------+--------+------------------+
-| NULL| NULL| 50.84999910990397|
-|Shanghai| w1|113.01666768391927|
+| Beijing| w1|103.73750019073486|
| Beijing| w2| 104.4000004359654|
+|Shanghai| w1|113.01666768391927|
|Shanghai| w2|100.10000038146973|
-| Beijing| w1|103.73750019073486|
+| null| null| 50.84999910990397|
+--------+--------+------------------+
Total line number = 5
It costs 0.027s
@@ -588,31 +588,62 @@ SELECT AVG(temperature) FROM root.factory1.** GROUP BY ([1000, 10000), 5s), TAGS
+-----------------------------+--------+--------+------------------+
| 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| w1|103.81666692097981|
|1970-01-01T08:00:01.000+08:00| Beijing| w2| 103.4|
+|1970-01-01T08:00:01.000+08:00|Shanghai| w1|113.20000076293945|
|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:01.000+08:00| null| null| 50.91999893188476|
+|1970-01-01T08:00:06.000+08:00| Beijing| w1| 103.5|
|1970-01-01T08:00:06.000+08:00| Beijing| w2| 106.9000015258789|
+|1970-01-01T08:00:06.000+08:00|Shanghai| w1| 112.6500015258789|
|1970-01-01T08:00:06.000+08:00|Shanghai| w2| 99.80000305175781|
-|1970-01-01T08:00:06.000+08:00| Beijing| w1| 103.5|
+|1970-01-01T08:00:06.000+08:00| null| null| 50.5|
+-----------------------------+--------+--------+------------------+
```
和标签聚合相比，基于时间区间的标签聚合的查询会首先按照时间区间划定聚合范围，在时间区间内部再根据指定的标签顺序，进行相应数据的聚合计算。在输出的结果集中，会包含一列时间列，该时间列值的含义和时间区间聚合查询的相同。
+### 标签聚合查询输出结果顺序
+
+在同聚合时间区间中，按照 `GROUP BY TAGS` 中指定的标签键依次比较，输出顺序为字符串的比较顺序。
+字符串的比较方式通过 Java 中的 [`String::compareTo`](https://docs.oracle.com/javase/8/docs/api/java/lang/String.html#compareTo-java.lang.String-) 实现。
+使用基于时间区间的聚合查询时，可以通过 `ORDER BY TIME` 指定时间顺序，时间区间按照指定的时间顺序输出，时间区间内仍然按照标签比较顺序输出。
+
+例如，上面的查询可以按照时间序列倒序输出
+
+SQL 语句如下
+
+```SQL
+SELECT AVG(temperature) FROM root.factory1.** GROUP BY ([1000, 10000), 5s), TAGS(city, workshop) ORDER BY TIME DESC;
+```
+
+查询结果如下
+
+```
++-----------------------------+--------+--------+------------------+
+| Time| city|workshop| avg(temperature)|
++-----------------------------+--------+--------+------------------+
+|1970-01-01T08:00:06.000+08:00| Beijing| w1| 103.5|
+|1970-01-01T08:00:06.000+08:00| Beijing| w2| 106.9000015258789|
+|1970-01-01T08:00:06.000+08:00|Shanghai| w1| 112.6500015258789|
+|1970-01-01T08:00:06.000+08:00|Shanghai| w2| 99.80000305175781|
+|1970-01-01T08:00:06.000+08:00| null| null| 50.5|
+|1970-01-01T08:00:01.000+08:00| Beijing| w1|103.81666692097981|
+|1970-01-01T08:00:01.000+08:00| Beijing| w2| 103.4|
+|1970-01-01T08:00:01.000+08:00|Shanghai| w1|113.20000076293945|
+|1970-01-01T08:00:01.000+08:00|Shanghai| w2| 100.1999994913737|
+|1970-01-01T08:00:01.000+08:00| null| null| 50.91999893188476|
++-----------------------------+--------+--------+------------------+
+```
### 标签聚合查询的限制
由于标签聚合功能仍然处于开发阶段，目前有如下未实现功能。
> 1. 暂不支持 `HAVING` 子句过滤查询结果。
> 2. 暂不支持结果按照标签值排序。
-> 3. 暂不支持 `LIMIT`，`OFFSET`，`SLIMIT`，`SOFFSET`。
-> 4. 暂不支持 `ALIGN BY DEVICE`。
-> 5. 暂不支持聚合函数内部包含表达式，例如 `count(s+1)`。
-> 6. 不支持值过滤条件聚合，和分层聚合查询行为保持一致。
+> 3. 暂不支持 `ALIGN BY DEVICE`。
+> 4. 暂不支持聚合函数内部包含表达式，例如 `count(s+1)`。
+> 5. 不支持值过滤条件聚合，和分层聚合查询行为保持一致。
## 聚合结果过滤
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
index c3a6989a1b..5ecd93c286 100644
--- 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
@@ -33,10 +33,6 @@ 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;
@@ -102,11 +98,11 @@ public class IoTDBTagAggregationIT {
// | 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|
+ // |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();
@@ -114,15 +110,6 @@ public class IoTDBTagAggregationIT {
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);
@@ -132,7 +119,16 @@ public class IoTDBTagAggregationIT {
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("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.assertNull(resultSet.getString(1));
Assert.assertEquals(4L, resultSet.getLong(2));
Assert.assertEquals(4.9D, resultSet.getDouble(3), DELTA);
Assert.assertEquals(10L, resultSet.getLong(4));
@@ -156,24 +152,15 @@ public class IoTDBTagAggregationIT {
// +----+------------+------------------+
// | k1|count(t + 1)| avg(t + 1)|
// +----+------------+------------------+
- // |k1v2| 4|3.1000000536441803|
// |k1v1| 6| 3.600000003973643|
- // |NULL| 4| 5.89999994635582|
+ // |k1v2| 4|3.1000000536441803|
+ // |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);
@@ -183,7 +170,16 @@ public class IoTDBTagAggregationIT {
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("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.assertNull(resultSet.getString(1));
Assert.assertEquals(4L, resultSet.getLong(2));
Assert.assertEquals(4.9D, resultSet.getDouble(3), DELTA);
Assert.assertEquals(10L, resultSet.getLong(4));
@@ -211,7 +207,7 @@ public class IoTDBTagAggregationIT {
// +----+--------+------------------+-----------+-----------+------------+------------+----------+
// |k1v2| 4|3.1000000536441803| 10| 1| 5.4| 1.3|
// 5.4|
- // |NULL| 4| 4.89999994635582| 10| 1| 8.7| 1.6|
+ // |null| 4| 4.89999994635582| 10| 1| 8.7| 1.6|
// 8.7|
// +----+--------+------------------+-----------+-----------+------------+------------+----------+
try (Connection connection = EnvFactory.getEnv().getConnection();
@@ -228,7 +224,7 @@ public class IoTDBTagAggregationIT {
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.assertNull(resultSet.getString(1));
Assert.assertEquals(4L, resultSet.getLong(2));
Assert.assertEquals(4.9D, resultSet.getDouble(3), DELTA);
Assert.assertEquals(10L, resultSet.getLong(4));
@@ -251,29 +247,35 @@ public class IoTDBTagAggregationIT {
// +----+----+--------+
// | k1| k2|count(t)|
// +----+----+--------+
- // |NULL|NULL| 2|
- // |NULL|k2v1| 2|
- // |k1v1|NULL| 2|
- // |k1v2|k2v1| 2|
- // |k1v1|k2v2| 2|
// |k1v1|k2v1| 2|
+ // |k1v1|k2v2| 2|
+ // |k1v1|null| 2|
+ // |k1v2|k2v1| 2|
// |k1v2|k2v2| 2|
+ // |null|k2v1| 2|
+ // |null|null| 2|
// +----+----+--------+
+ Object[][] expected =
+ new Object[][] {
+ {"k1v1", "k2v1", 2L},
+ {"k1v1", "k2v2", 2L},
+ {"k1v1", null, 2L},
+ {"k1v2", "k2v1", 2L},
+ {"k1v2", "k2v2", 2L},
+ {null, "k2v1", 2L},
+ {null, null, 2L},
+ };
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++) {
+ for (Object[] objects : expected) {
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.assertEquals(objects[0], resultSet.getString("k1"));
+ Assert.assertEquals(objects[1], resultSet.getString("k2"));
+ Assert.assertEquals(objects[2], resultSet.getLong("count(t)"));
}
Assert.assertFalse(resultSet.next());
- Assert.assertEquals(7, groups.size());
}
} catch (SQLException e) {
e.printStackTrace();
@@ -288,42 +290,32 @@ public class IoTDBTagAggregationIT {
// +-----------------------------+----+--------+
// | 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.000+08:00|k1v2| 2|
+ // |1970-01-01T08:00:00.000+08:00|null| 2|
// |1970-01-01T08:00:00.010+08:00|k1v1| 3|
- // |1970-01-01T08:00:00.010+08:00|NULL| 2|
+ // |1970-01-01T08:00:00.010+08:00|k1v2| 2|
+ // |1970-01-01T08:00:00.010+08:00|null| 2|
// +-----------------------------+----+--------+
+ Object[][] expected =
+ new Object[][] {
+ {0L, "k1v1", 3L},
+ {0L, "k1v2", 2L},
+ {0L, null, 2L},
+ {10L, "k1v1", 3L},
+ {10L, "k1v2", 2L},
+ {10L, null, 2L},
+ };
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++) {
+ for (Object[] objects : expected) {
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(objects[0], resultSet.getLong("Time"));
+ Assert.assertEquals(objects[1], resultSet.getString("k1"));
+ Assert.assertEquals(objects[2], resultSet.getLong("count(t)"));
}
- Assert.assertEquals(3, groups.size());
Assert.assertFalse(resultSet.next());
}
} catch (SQLException e) {
@@ -339,48 +331,48 @@ public class IoTDBTagAggregationIT {
// +-----------------------------+----+--------+
// | Time| k1|count(t)|
// +-----------------------------+----+--------+
- // |1970-01-01T08:00:00.000+08:00|k1v2| 4|
// |1970-01-01T08:00:00.000+08:00|k1v1| 6|
- // |1970-01-01T08:00:00.000+08:00|NULL| 4|
- // |1970-01-01T08:00:00.005+08:00|k1v2| 2|
+ // |1970-01-01T08:00:00.000+08:00|k1v2| 4|
+ // |1970-01-01T08:00:00.000+08:00|null| 4|
// |1970-01-01T08:00:00.005+08:00|k1v1| 3|
- // |1970-01-01T08:00:00.005+08:00|NULL| 2|
- // |1970-01-01T08:00:00.010+08:00|k1v2| 2|
+ // |1970-01-01T08:00:00.005+08:00|k1v2| 2|
+ // |1970-01-01T08:00:00.005+08:00|null| 2|
// |1970-01-01T08:00:00.010+08:00|k1v1| 3|
- // |1970-01-01T08:00:00.010+08:00|NULL| 2|
- // |1970-01-01T08:00:00.015+08:00|k1v2| 0|
+ // |1970-01-01T08:00:00.010+08:00|k1v2| 2|
+ // |1970-01-01T08:00:00.010+08:00|null| 2|
// |1970-01-01T08:00:00.015+08:00|k1v1| 0|
- // |1970-01-01T08:00:00.015+08:00|NULL| 0|
+ // |1970-01-01T08:00:00.015+08:00|k1v2| 0|
+ // |1970-01-01T08:00:00.015+08:00|null| 0|
// +-----------------------------+----+--------+
- long[][] expectedValue = new long[][] {{4L, 6L, 4L}, {2L, 3L, 2L}, {2L, 3L, 2L}, {0L, 0L, 0L}};
- long[] expectedTime = new long[] {0L, 5L, 10L, 15L};
+ Object[][] expected =
+ new Object[][] {
+ {0L, "k1v1", 6L},
+ {0L, "k1v2", 4L},
+ {0L, null, 4L},
+ {5L, "k1v1", 3L},
+ {5L, "k1v2", 2L},
+ {5L, null, 2L},
+ {10L, "k1v1", 3L},
+ {10L, "k1v2", 2L},
+ {10L, null, 2L},
+ {15L, "k1v1", 0L},
+ {15L, "k1v2", 0L},
+ {15L, null, 0L},
+ };
try (Connection connection = EnvFactory.getEnv().getConnection();
Statement statement = connection.createStatement()) {
try (ResultSet resultSet = statement.executeQuery(query)) {
Assert.assertEquals(3, resultSet.getMetaData().getColumnCount());
- for (int i = 0; i < 4; i++) {
- for (int j = 0; j < 3; j++) {
- Assert.assertTrue(resultSet.next());
- String tagValue = resultSet.getString("k1");
- switch (tagValue) {
- case "k1v2":
- Assert.assertEquals(expectedTime[i], resultSet.getLong("Time"));
- Assert.assertEquals(expectedValue[i][0], resultSet.getLong("count(t)"));
- break;
- case "k1v1":
- Assert.assertEquals(expectedValue[i][1], resultSet.getLong("count(t)"));
- break;
- case "NULL":
- Assert.assertEquals(expectedValue[i][2], resultSet.getLong("count(t)"));
- break;
- default:
- fail("Unexpected tag value: " + tagValue);
- }
- }
+ for (Object[] objects : expected) {
+ Assert.assertTrue(resultSet.next());
+ Assert.assertEquals(objects[0], resultSet.getLong("Time"));
+ Assert.assertEquals(objects[1], resultSet.getString("k1"));
+ Assert.assertEquals(objects[2], resultSet.getLong("count(t)"));
}
Assert.assertFalse(resultSet.next());
}
} catch (SQLException e) {
+ e.printStackTrace();
fail(e.getMessage());
}
}
@@ -393,42 +385,32 @@ public class IoTDBTagAggregationIT {
// +-----------------------------+----+--------+
// | Time| k1|count(t)|
// +-----------------------------+----+--------+
- // |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|k1v2| 2|
// |1970-01-01T08:00:00.010+08:00|NULL| 2|
- // |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|k1v2| 2|
// |1970-01-01T08:00:00.000+08:00|NULL| 2|
// +-----------------------------+----+--------+
+ Object[][] expected =
+ new Object[][] {
+ {10L, "k1v1", 3L},
+ {10L, "k1v2", 2L},
+ {10L, null, 2L},
+ {0L, "k1v1", 3L},
+ {0L, "k1v2", 2L},
+ {0L, null, 2L},
+ };
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++) {
+ for (Object[] objects : expected) {
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(objects[0], resultSet.getLong("Time"));
+ Assert.assertEquals(objects[1], resultSet.getString("k1"));
+ Assert.assertEquals(objects[2], resultSet.getLong("count(t)"));
}
- Assert.assertEquals(3, groups.size());
Assert.assertFalse(resultSet.next());
}
} catch (SQLException e) {
@@ -444,36 +426,26 @@ public class IoTDBTagAggregationIT {
// +----+--------+
// | k1|count(t)|
// +----+--------+
- // |k1v2| 2|
// |k1v1| 3|
+ // |k1v2| 2|
// |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.assertTrue(resultSet.next());
+ Assert.assertEquals("k1v1", resultSet.getString("k1"));
+ Assert.assertEquals(3, resultSet.getLong("count(t)"));
+ Assert.assertTrue(resultSet.next());
+ Assert.assertEquals("k1v2", resultSet.getString("k1"));
+ Assert.assertEquals(2, resultSet.getLong("count(t)"));
+ Assert.assertTrue(resultSet.next());
+ Assert.assertNull(resultSet.getString("k1"));
+ Assert.assertEquals(2, resultSet.getLong("count(t)"));
Assert.assertFalse(resultSet.next());
}
+
} catch (SQLException e) {
e.printStackTrace();
fail(e.getMessage());
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
index 4abbd1dcb3..8ebc3716c5 100644
--- 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
@@ -151,7 +151,7 @@ public class TagAggregationOperator implements ProcessOperator {
for (int i = 0; i < group.size(); i++) {
if (group.get(i) == null) {
- columnBuilders[i].writeBinary(new Binary("NULL"));
+ columnBuilders[i].appendNull();
} else {
columnBuilders[i].writeBinary(new Binary(group.get(i)));
}
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 965c4bd477..3fdfa3774b 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
@@ -98,6 +98,8 @@ import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
+import java.util.SortedMap;
+import java.util.TreeMap;
import java.util.stream.Collectors;
import static com.google.common.base.Preconditions.checkArgument;
@@ -650,8 +652,8 @@ public class LogicalPlanBuilder {
AggregationStep curStep,
GroupByTimeParameter groupByTimeParameter,
Ordering scanOrder) {
- Map<List<String>, List<CrossSeriesAggregationDescriptor>> tagValuesToAggregationDescriptors =
- new HashMap<>();
+ SortedMap<List<String>, List<CrossSeriesAggregationDescriptor>>
+ tagValuesToAggregationDescriptors = new TreeMap<>(GroupByTagNode::tagValuesComparator);
for (List<String> tagValues : tagValuesToGroupedTimeseriesOperands.keySet()) {
LinkedHashMap<Expression, List<Expression>> groupedTimeseriesOperands =
tagValuesToGroupedTimeseriesOperands.get(tagValues);
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
index e0ef548508..c8c03b024d 100644
--- 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
@@ -36,18 +36,23 @@ import java.io.DataOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
-import java.util.HashMap;
import java.util.List;
-import java.util.Map;
import java.util.Map.Entry;
import java.util.Objects;
+import java.util.SortedMap;
+import java.util.TreeMap;
import java.util.stream.Collectors;
public class GroupByTagNode extends MultiChildNode {
private final List<String> tagKeys;
- private final Map<List<String>, List<CrossSeriesAggregationDescriptor>>
+
+ /**
+ * the tag values should be in a deterministic order so that the output is deterministic as well.
+ */
+ private final SortedMap<List<String>, List<CrossSeriesAggregationDescriptor>>
tagValuesToAggregationDescriptors;
+
private final List<String> outputColumnNames;
// The parameter of `group by time`.
@@ -62,7 +67,8 @@ public class GroupByTagNode extends MultiChildNode {
@Nullable GroupByTimeParameter groupByTimeParameter,
Ordering scanOrder,
List<String> tagKeys,
- Map<List<String>, List<CrossSeriesAggregationDescriptor>> tagValuesToAggregationDescriptors,
+ SortedMap<List<String>, List<CrossSeriesAggregationDescriptor>>
+ tagValuesToAggregationDescriptors,
List<String> outputColumnNames) {
super(id, children);
this.groupByTimeParameter = groupByTimeParameter;
@@ -77,7 +83,8 @@ public class GroupByTagNode extends MultiChildNode {
@Nullable GroupByTimeParameter groupByTimeParameter,
Ordering scanOrder,
List<String> tagKeys,
- Map<List<String>, List<CrossSeriesAggregationDescriptor>> tagValuesToAggregationDescriptors,
+ SortedMap<List<String>, List<CrossSeriesAggregationDescriptor>>
+ tagValuesToAggregationDescriptors,
List<String> outputColumnNames) {
super(id);
this.groupByTimeParameter = groupByTimeParameter;
@@ -217,7 +224,7 @@ public class GroupByTagNode extends MultiChildNode {
return tagKeys;
}
- public Map<List<String>, List<CrossSeriesAggregationDescriptor>>
+ public SortedMap<List<String>, List<CrossSeriesAggregationDescriptor>>
getTagValuesToAggregationDescriptors() {
return tagValuesToAggregationDescriptors;
}
@@ -228,8 +235,8 @@ public class GroupByTagNode extends MultiChildNode {
// Tag values to aggregation descriptors.
int numOfEntries = ReadWriteIOUtils.readInt(byteBuffer);
- Map<List<String>, List<CrossSeriesAggregationDescriptor>> tagValuesToAggregationDescriptors =
- new HashMap<>();
+ SortedMap<List<String>, List<CrossSeriesAggregationDescriptor>>
+ tagValuesToAggregationDescriptors = new TreeMap<>(GroupByTagNode::tagValuesComparator);
while (numOfEntries > 0) {
List<String> tagValues = ReadWriteIOUtils.readStringList(byteBuffer);
List<CrossSeriesAggregationDescriptor> aggregationDescriptors = new ArrayList<>();
@@ -309,4 +316,40 @@ public class GroupByTagNode extends MultiChildNode {
list -> list.stream().map(CrossSeriesAggregationDescriptor::getInputExpressions))
.collect(Collectors.toList()));
}
+
+ /**
+ * The comparator of tag values in GROUP BY tag query, it uses the following rules to sort:
+ *
+ * <p>1. As each group of tag values must have the same length, so there's no need to compare the
+ * size of the lists.
+ *
+ * <p>2. Let v1(i) be the ith element in v1, then v1 < v2 iff there exists any i (0 <= i <
+ * len(v1)) that
+ *
+ * <ul>
+ * <li>a. v1(j) == v2(j) when 0 <= j < i, and
+ * <li>b. v1(i) < v2(i) in natual order of String, or
+ * <ul>
+ * <li>b1. v1(i) != null and v2(i) == null.
+ * </ul>
+ * </ul>
+ *
+ * <p>The NULL value will be output in the last of every in-series aggregation group.
+ */
+ public static int tagValuesComparator(List<String> v1, List<String> v2) {
+ Validate.isTrue(v1.size() == v2.size());
+ for (int i = 0; i < v1.size(); i++) {
+ if (Objects.equals(v1.get(i), v2.get(i))) {
+ continue;
+ }
+ if (v2.get(i) == null) {
+ return -1;
+ }
+ if (v1.get(i) == null) {
+ return 1;
+ }
+ return v1.get(i).compareTo(v2.get(i));
+ }
+ return 0;
+ }
}
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
index feaaf4fcaa..378c1facd2 100644
--- 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
@@ -34,6 +34,7 @@ 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.apache.iotdb.tsfile.utils.Pair;
import org.junit.Assert;
import org.junit.Test;
@@ -44,10 +45,10 @@ 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;
+import java.util.SortedMap;
+import java.util.TreeMap;
public class GroupByTagNodeSerdeTest {
@@ -85,8 +86,8 @@ public class GroupByTagNodeSerdeTest {
AggregationType.AVG.name().toLowerCase(),
AggregationStep.PARTIAL,
Collections.singletonList(new TimeSeriesOperand(new PartialPath("root.sg.d1.s1"))));
- Map<List<String>, List<CrossSeriesAggregationDescriptor>> tagValuesToAggregationDescriptors =
- new HashMap<>();
+ SortedMap<List<String>, List<CrossSeriesAggregationDescriptor>>
+ tagValuesToAggregationDescriptors = new TreeMap<>(GroupByTagNode::tagValuesComparator);
tagValuesToAggregationDescriptors.put(
Arrays.asList("v1", "v2"), Arrays.asList(s1MaxTime, s1Avg));
GroupByTagNode expectedNode =
@@ -119,4 +120,34 @@ public class GroupByTagNodeSerdeTest {
ByteBuffer buffer = ByteBuffer.wrap(byteArray);
Assert.assertEquals(expectedNode, PlanNodeDeserializeHelper.deserialize(buffer));
}
+
+ @Test
+ public void testTagValueComparator() {
+ List<Pair<List<String>, List<String>>> testCases =
+ Arrays.asList(
+ new Pair<>(Arrays.asList("x", "y", "z"), Arrays.asList("A", "B", "C")),
+ new Pair<>(Arrays.asList("x", "y", "z"), Arrays.asList("x", null, "z")),
+ new Pair<>(Arrays.asList(null, "y", "z"), Arrays.asList("x", null, null)),
+ new Pair<>(Arrays.asList(null, null, null), Arrays.asList(null, null, null)),
+ new Pair<>(Arrays.asList("x", "y", "z"), Arrays.asList("x", "y", "z")),
+ new Pair<>(Arrays.asList("x", "yz"), Arrays.asList("x", "y")));
+ int[] expected = new int[] {1, -1, 1, 0, 0, 1};
+ for (int i = 0; i < testCases.size(); i++) {
+ Pair<List<String>, List<String>> testCase = testCases.get(i);
+ if (expected[i] == 0) {
+ Assert.assertEquals(
+ testCase.toString(),
+ expected[i],
+ GroupByTagNode.tagValuesComparator(testCase.left, testCase.right));
+ } else if (expected[i] > 0) {
+ Assert.assertTrue(
+ testCase.toString(),
+ GroupByTagNode.tagValuesComparator(testCase.left, testCase.right) > 0);
+ } else {
+ Assert.assertTrue(
+ testCase.toString(),
+ GroupByTagNode.tagValuesComparator(testCase.left, testCase.right) < 0);
+ }
+ }
+ }
}