You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ro...@apache.org on 2021/12/11 14:36:11 UTC
[iotdb] branch master updated: [IOTDB-2092] Basic support aggregation query with UDF nested (#4517)
This is an automated email from the ASF dual-hosted git repository.
rong pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 5483abf [IOTDB-2092] Basic support aggregation query with UDF nested (#4517)
5483abf is described below
commit 5483abf4c22e285eeeaeef49a40fd191d29194f1
Author: BaiJian <er...@hotmail.com>
AuthorDate: Sat Dec 11 22:35:37 2021 +0800
[IOTDB-2092] Basic support aggregation query with UDF nested (#4517)
---
.../DML-Data-Manipulation-Language.md | 77 +++
.../DML-Data-Manipulation-Language.md | 78 ++++
.../aggregation/IoTDBUDFNestAggregationIT.java | 515 +++++++++++++++++++++
.../apache/iotdb/db/qp/executor/PlanExecutor.java | 4 +
.../qp/logical/crud/AggregationQueryOperator.java | 38 +-
.../db/qp/logical/crud/FillQueryOperator.java | 2 +-
.../iotdb/db/qp/logical/crud/QueryOperator.java | 29 +-
.../iotdb/db/qp/logical/crud/SelectComponent.java | 12 +-
.../db/qp/logical/crud/UDAFQueryOperator.java | 172 +++++++
.../apache/iotdb/db/qp/physical/crud/UDAFPlan.java | 68 +++
.../apache/iotdb/db/qp/physical/crud/UDTFPlan.java | 6 +-
.../apache/iotdb/db/qp/sql/IoTDBSqlVisitor.java | 30 +-
.../iotdb/db/qp/utils/GroupByLevelController.java | 83 ++--
...{UDFInputDataSet.java => IUDFInputDataSet.java} | 10 +-
.../dataset/RawQueryDataSetWithValueFilter.java | 2 +-
.../iotdb/db/query/dataset/UDFInputDataSet.java | 42 +-
.../UDFRawQueryInputDataSetWithoutValueFilter.java | 2 +-
.../db/query/dataset/UDTFAlignByTimeDataSet.java | 184 ++++----
.../apache/iotdb/db/query/dataset/UDTFDataSet.java | 8 +
.../db/query/dataset/UDTFNonAlignDataSet.java | 2 +-
.../iotdb/db/query/executor/IQueryRouter.java | 6 +
.../iotdb/db/query/executor/QueryRouter.java | 35 +-
...DTFQueryExecutor.java => UDFQueryExecutor.java} | 16 +-
.../iotdb/db/query/expression/Expression.java | 46 ++
.../iotdb/db/query/expression/ResultColumn.java | 17 +
.../query/expression/binary/BinaryExpression.java | 24 +-
.../db/query/expression/unary/ConstantOperand.java | 6 +
.../query/expression/unary/FunctionExpression.java | 53 ++-
.../query/expression/unary/NegationExpression.java | 11 +
.../query/expression/unary/TimeSeriesOperand.java | 6 +
.../db/query/udf/core/executor/UDTFExecutor.java | 13 +-
.../core/layer/InputRowUtils.java} | 33 +-
.../db/query/udf/core/layer/LayerCacheUtils.java | 53 ++-
.../layer/MultiInputColumnIntermediateLayer.java | 65 +--
.../query/udf/core/layer/RawQueryInputLayer.java | 25 +-
...InputColumnMultiReferenceIntermediateLayer.java | 10 +
...nputColumnSingleReferenceIntermediateLayer.java | 11 +-
.../udf/core/reader/ConstantLayerPointReader.java | 5 +
.../db/query/udf/core/reader/LayerPointReader.java | 2 +
.../db/query/udf/core/reader/LayerRowReader.java | 3 +
.../transformer/ArithmeticBinaryTransformer.java | 12 +-
.../transformer/ArithmeticNegationTransformer.java | 36 +-
.../db/query/udf/core/transformer/Transformer.java | 10 +-
.../core/transformer/TransparentTransformer.java | 93 ++++
.../core/transformer/UDFQueryRowTransformer.java | 5 +-
.../udf/core/transformer/UDFQueryTransformer.java | 14 +-
.../row/ElasticSerializableRowRecordList.java | 30 +-
.../row/SerializableRowRecordList.java | 3 +-
.../tv/ElasticSerializableBinaryTVList.java | 8 +-
.../tv/ElasticSerializableTVList.java | 47 ++
.../datastructure/tv/SerializableBinaryTVList.java | 5 +-
.../tv/SerializableBooleanTVList.java | 5 +-
.../datastructure/tv/SerializableDoubleTVList.java | 5 +-
.../datastructure/tv/SerializableFloatTVList.java | 5 +-
.../datastructure/tv/SerializableIntTVList.java | 3 +-
.../datastructure/tv/SerializableLongTVList.java | 5 +-
.../db/service/thrift/impl/TSServiceImpl.java | 93 +++-
.../ElasticSerializableRowRecordListTest.java | 127 +++--
.../ElasticSerializableTVListTest.java | 175 +++++--
.../iotdb/tsfile/utils/ReadWriteIOUtils.java | 3 +-
60 files changed, 2118 insertions(+), 370 deletions(-)
diff --git a/docs/UserGuide/IoTDB-SQL-Language/DML-Data-Manipulation-Language.md b/docs/UserGuide/IoTDB-SQL-Language/DML-Data-Manipulation-Language.md
index 8ea04b1..0d01221 100644
--- a/docs/UserGuide/IoTDB-SQL-Language/DML-Data-Manipulation-Language.md
+++ b/docs/UserGuide/IoTDB-SQL-Language/DML-Data-Manipulation-Language.md
@@ -1014,6 +1014,83 @@ Total line number = 7
It costs 0.004s
```
+#### Nested Expressions query with aggregations
+
+IoTDB supports aggregation query nested by any other expressions.
+
+##### Example
+
+1. Aggregation query without `GROUP BY`.
+
+Input:
+
+```sql
+select avg(temperature),
+ sin(avg(temperature)),
+ avg(temperature) + 1,
+ -sum(hardware),
+ avg(temperature) + sum(hardware)
+from root.ln.wf01.wt01;
+```
+
+Result:
+
+```
++----------------------------------+---------------------------------------+--------------------------------------+--------------------------------+--------------------------------------------------------------------+
+|avg(root.ln.wf01.wt01.temperature)|sin(avg(root.ln.wf01.wt01.temperature))|avg(root.ln.wf01.wt01.temperature) + 1|-sum(root.ln.wf01.wt01.hardware)|avg(root.ln.wf01.wt01.temperature) + sum(root.ln.wf01.wt01.hardware)|
++----------------------------------+---------------------------------------+--------------------------------------+--------------------------------+--------------------------------------------------------------------+
+| 15.927999999999999| -0.21826546964855045| 16.927999999999997| -7426.0| 7441.928|
++----------------------------------+---------------------------------------+--------------------------------------+--------------------------------+--------------------------------------------------------------------+
+Total line number = 1
+It costs 0.009s
+```
+
+2. Aggregation with `GROUP BY`.
+
+Input:
+
+```sql
+select avg(temperature),
+ sin(avg(temperature)),
+ avg(temperature) + 1,
+ -sum(hardware),
+ avg(temperature) + sum(hardware) as custom_sum
+from root.ln.wf01.wt01
+GROUP BY([10, 90), 10ms);
+```
+
+Result:
+
+```
++-----------------------------+----------------------------------+---------------------------------------+--------------------------------------+--------------------------------+----------+
+| Time|avg(root.ln.wf01.wt01.temperature)|sin(avg(root.ln.wf01.wt01.temperature))|avg(root.ln.wf01.wt01.temperature) + 1|-sum(root.ln.wf01.wt01.hardware)|custom_sum|
++-----------------------------+----------------------------------+---------------------------------------+--------------------------------------+--------------------------------+----------+
+|1970-01-01T08:00:00.010+08:00| 13.987499999999999| 0.9888207947857667| 14.987499999999999| -3211.0| 3224.9875|
+|1970-01-01T08:00:00.020+08:00| 29.6| -0.9701057337071853| 30.6| -3720.0| 3749.6|
+|1970-01-01T08:00:00.030+08:00| null| null| null| null| null|
+|1970-01-01T08:00:00.040+08:00| null| null| null| null| null|
+|1970-01-01T08:00:00.050+08:00| null| null| null| null| null|
+|1970-01-01T08:00:00.060+08:00| null| null| null| null| null|
+|1970-01-01T08:00:00.070+08:00| null| null| null| null| null|
+|1970-01-01T08:00:00.080+08:00| null| null| null| null| null|
++-----------------------------+----------------------------------+---------------------------------------+--------------------------------------+--------------------------------+----------+
+Total line number = 8
+It costs 0.012s
+```
+
+##### Note
+
+> Automated fill (`FILL`) and grouped by level (`GROUP BY LEVEL`) are not supported in an aggregation query with expression nested. They may be supported in future versions.
+>
+> The aggregation expression must be the lowest level input of one expression tree. Any kind expressions except timeseries are not valid as aggregation function parameters。
+>
+> In a word, the following queries are not valid.
+> ```sql
+> SELECT avg(s1+1) FROM root.sg.d1; -- The aggregation function has expression parameters.
+> SELECT avg(s1) + avg(s2) FROM root.sg.* GROUP BY LEVEL=1; -- Grouped by level
+> SELECT avg(s1) + avg(s2) FROM root.sg.d1 GROUP BY([0, 10000), 1s) FILL(double [previous]); -- Automated fill
+> ```
+
### Automated Fill
In the actual use of IoTDB, when doing the query operation of timeseries, situations where the value is null at some time points may appear, which will obstruct the further analysis by users. In order to better reflect the degree of data change, users expect missing values to be automatically filled. Therefore, the IoTDB system introduces the function of Automated Fill.
diff --git a/docs/zh/UserGuide/IoTDB-SQL-Language/DML-Data-Manipulation-Language.md b/docs/zh/UserGuide/IoTDB-SQL-Language/DML-Data-Manipulation-Language.md
index 3ffa833..5759925 100644
--- a/docs/zh/UserGuide/IoTDB-SQL-Language/DML-Data-Manipulation-Language.md
+++ b/docs/zh/UserGuide/IoTDB-SQL-Language/DML-Data-Manipulation-Language.md
@@ -1019,6 +1019,84 @@ select count(status) from root.ln.wf01.wt01 group by ([0,20),2ms,3ms), level=1;
Total line number = 7
It costs 0.004s
```
+
+#### 聚合查询嵌套表达式
+
+IoTDB 支持在 `SELECT` 字句中执行由聚合查询和其他运算组成的任意嵌套表达式。
+
+##### 示例
+
+不指定 `GROUP BY` 的聚合查询。
+
+输入：
+
+```sql
+select avg(temperature),
+ sin(avg(temperature)),
+ avg(temperature) + 1,
+ -sum(hardware),
+ avg(temperature) + sum(hardware)
+from root.ln.wf01.wt01;
+```
+
+结果：
+
+```
++----------------------------------+---------------------------------------+--------------------------------------+--------------------------------+--------------------------------------------------------------------+
+|avg(root.ln.wf01.wt01.temperature)|sin(avg(root.ln.wf01.wt01.temperature))|avg(root.ln.wf01.wt01.temperature) + 1|-sum(root.ln.wf01.wt01.hardware)|avg(root.ln.wf01.wt01.temperature) + sum(root.ln.wf01.wt01.hardware)|
++----------------------------------+---------------------------------------+--------------------------------------+--------------------------------+--------------------------------------------------------------------+
+| 15.927999999999999| -0.21826546964855045| 16.927999999999997| -7426.0| 7441.928|
++----------------------------------+---------------------------------------+--------------------------------------+--------------------------------+--------------------------------------------------------------------+
+Total line number = 1
+It costs 0.009s
+```
+
+指定 `GROUP BY` 的聚合查询。
+
+输入：
+
+```sql
+select avg(temperature),
+ sin(avg(temperature)),
+ avg(temperature) + 1,
+ -sum(hardware),
+ avg(temperature) + sum(hardware) as custom_sum
+from root.ln.wf01.wt01
+GROUP BY([10, 90), 10ms);
+```
+
+结果：
+
+```
++-----------------------------+----------------------------------+---------------------------------------+--------------------------------------+--------------------------------+----------+
+| Time|avg(root.ln.wf01.wt01.temperature)|sin(avg(root.ln.wf01.wt01.temperature))|avg(root.ln.wf01.wt01.temperature) + 1|-sum(root.ln.wf01.wt01.hardware)|custom_sum|
++-----------------------------+----------------------------------+---------------------------------------+--------------------------------------+--------------------------------+----------+
+|1970-01-01T08:00:00.010+08:00| 13.987499999999999| 0.9888207947857667| 14.987499999999999| -3211.0| 3224.9875|
+|1970-01-01T08:00:00.020+08:00| 29.6| -0.9701057337071853| 30.6| -3720.0| 3749.6|
+|1970-01-01T08:00:00.030+08:00| null| null| null| null| null|
+|1970-01-01T08:00:00.040+08:00| null| null| null| null| null|
+|1970-01-01T08:00:00.050+08:00| null| null| null| null| null|
+|1970-01-01T08:00:00.060+08:00| null| null| null| null| null|
+|1970-01-01T08:00:00.070+08:00| null| null| null| null| null|
+|1970-01-01T08:00:00.080+08:00| null| null| null| null| null|
++-----------------------------+----------------------------------+---------------------------------------+--------------------------------------+--------------------------------+----------+
+Total line number = 8
+It costs 0.012s
+```
+
+##### 注意
+
+> 目前此功能尚不支持填充算子（`FILL`）和按层级聚合（`GROUP BY LEVEL`）查询, 在后续版本会支持。
+>
+> 聚合计算目前只能当做最底层表达式输入，暂不支持聚合函数内部出现表达式。
+>
+> 即不支持以下查询
+> ```SQL
+> SELECT avg(s1+1) FROM root.sg.d1; -- 聚合函数内部有表达式
+> SELECT avg(s1) + avg(s2) FROM root.sg.* GROUP BY LEVEL=1; -- 按层级聚合
+> SELECT avg(s1) + avg(s2) FROM root.sg.d1 GROUP BY([0, 10000), 1s) FILL(double [previous]); -- 空值填充
+> ```
+
### 空值填充
在 IoTDB 的实际使用中，当进行时间序列的查询操作时，可能会出现在某些时间点值为 null 的情况，这会妨碍用户进行进一步的分析。 为了更好地反映数据更改的程度，用户希望可以自动填充缺失值。 因此，IoTDB 系统引入了自动填充功能。
diff --git a/integration/src/test/java/org/apache/iotdb/db/integration/aggregation/IoTDBUDFNestAggregationIT.java b/integration/src/test/java/org/apache/iotdb/db/integration/aggregation/IoTDBUDFNestAggregationIT.java
new file mode 100644
index 0000000..79a1155
--- /dev/null
+++ b/integration/src/test/java/org/apache/iotdb/db/integration/aggregation/IoTDBUDFNestAggregationIT.java
@@ -0,0 +1,515 @@
+/*
+ * 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.integration.aggregation;
+
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.integration.env.ConfigFactory;
+import org.apache.iotdb.integration.env.EnvFactory;
+import org.apache.iotdb.itbase.category.ClusterTest;
+import org.apache.iotdb.itbase.category.LocalStandaloneTest;
+
+import org.junit.AfterClass;
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+import java.sql.Connection;
+import java.sql.ResultSet;
+import java.sql.Statement;
+import java.util.Locale;
+
+import static org.junit.Assert.fail;
+
+@Category({LocalStandaloneTest.class, ClusterTest.class})
+public class IoTDBUDFNestAggregationIT {
+
+ private static final double E = 1e-6;
+ private static final String TIMESTAMP_STR = "Time";
+
+ private static String[] creationSqls =
+ new String[] {
+ "SET STORAGE GROUP TO root.vehicle.d0",
+ "SET STORAGE GROUP TO root.vehicle.d1",
+ "CREATE TIMESERIES root.vehicle.d0.s0 WITH DATATYPE=INT32, ENCODING=RLE",
+ "CREATE TIMESERIES root.vehicle.d0.s1 WITH DATATYPE=INT64, ENCODING=RLE",
+ "CREATE TIMESERIES root.vehicle.d0.s2 WITH DATATYPE=FLOAT, ENCODING=RLE",
+ "CREATE TIMESERIES root.vehicle.d0.s3 WITH DATATYPE=TEXT, ENCODING=PLAIN",
+ "CREATE TIMESERIES root.vehicle.d0.s4 WITH DATATYPE=BOOLEAN, ENCODING=PLAIN"
+ };
+ private static String[] dataSet2 =
+ new String[] {
+ "SET STORAGE GROUP TO root.ln.wf01.wt01",
+ "CREATE TIMESERIES root.ln.wf01.wt01.status WITH DATATYPE=BOOLEAN, ENCODING=PLAIN",
+ "CREATE TIMESERIES root.ln.wf01.wt01.temperature WITH DATATYPE=FLOAT, ENCODING=PLAIN",
+ "CREATE TIMESERIES root.ln.wf01.wt01.hardware WITH DATATYPE=INT32, ENCODING=PLAIN",
+ "INSERT INTO root.ln.wf01.wt01(timestamp,temperature,status, hardware) "
+ + "values(1, 1.1, false, 11)",
+ "INSERT INTO root.ln.wf01.wt01(timestamp,temperature,status, hardware) "
+ + "values(2, 2.2, true, 22)",
+ "INSERT INTO root.ln.wf01.wt01(timestamp,temperature,status, hardware) "
+ + "values(3, 3.3, false, 33 )",
+ "INSERT INTO root.ln.wf01.wt01(timestamp,temperature,status, hardware) "
+ + "values(4, 4.4, false, 44)",
+ "INSERT INTO root.ln.wf01.wt01(timestamp,temperature,status, hardware) "
+ + "values(5, 5.5, false, 55)"
+ };
+ private static String[] dataSet3 =
+ new String[] {
+ "SET STORAGE GROUP TO root.sg",
+ "CREATE TIMESERIES root.sg.d1.s1 WITH DATATYPE=INT32, ENCODING=RLE",
+ "insert into root.sg.d1(timestamp,s1) values(5,5)",
+ "insert into root.sg.d1(timestamp,s1) values(12,12)",
+ "flush",
+ "insert into root.sg.d1(timestamp,s1) values(15,15)",
+ "insert into root.sg.d1(timestamp,s1) values(25,25)",
+ "flush",
+ "insert into root.sg.d1(timestamp,s1) values(1,111)",
+ "insert into root.sg.d1(timestamp,s1) values(20,200)",
+ "flush"
+ };
+
+ private static final String[] dataSet4 =
+ new String[] {
+ "SET STORAGE GROUP TO root.sg1",
+ "SET STORAGE GROUP TO root.sg2",
+ "CREATE TIMESERIES root.sg1.d1.status WITH DATATYPE=BOOLEAN, ENCODING=PLAIN",
+ "CREATE TIMESERIES root.sg1.d1.temperature WITH DATATYPE=DOUBLE, ENCODING=PLAIN",
+ "CREATE TIMESERIES root.sg1.d2.status WITH DATATYPE=BOOLEAN, ENCODING=PLAIN",
+ "CREATE TIMESERIES root.sg1.d2.temperature WITH DATATYPE=DOUBLE, ENCODING=PLAIN",
+ "CREATE TIMESERIES root.sg2.d1.status WITH DATATYPE=INT32, ENCODING=PLAIN",
+ "CREATE TIMESERIES root.sg2.d1.temperature WITH DATATYPE=DOUBLE, ENCODING=PLAIN",
+ "CREATE TIMESERIES root.sg2.d2.temperature WITH DATATYPE=DOUBLE, ENCODING=PLAIN",
+ "INSERT INTO root.sg1.d1(timestamp,status) values(150,true)",
+ "INSERT INTO root.sg1.d1(timestamp,status,temperature) values(200,false,20.71)",
+ "INSERT INTO root.sg1.d1(timestamp,status,temperature) values(600,false,71.12)",
+ "INSERT INTO root.sg1.d2(timestamp,status,temperature) values(200,false,42.66)",
+ "INSERT INTO root.sg1.d2(timestamp,status,temperature) values(300,false,46.77)",
+ "INSERT INTO root.sg1.d2(timestamp,status,temperature) values(700,true,62.15)",
+ "INSERT INTO root.sg2.d1(timestamp,status,temperature) values(100,3,88.24)",
+ "INSERT INTO root.sg2.d1(timestamp,status,temperature) values(500,5,125.5)",
+ "INSERT INTO root.sg2.d2(timestamp,temperature) values(200,105.5)",
+ "INSERT INTO root.sg2.d2(timestamp,temperature) values(800,61.22)",
+ };
+
+ private final String d0s0 = "root.vehicle.d0.s0";
+ private final String d0s1 = "root.vehicle.d0.s1";
+ private final String d0s2 = "root.vehicle.d0.s2";
+ private final String d0s3 = "root.vehicle.d0.s3";
+ private static final String insertTemplate =
+ "INSERT INTO root.vehicle.d0(timestamp,s0,s1,s2,s3,s4)" + " VALUES(%d,%d,%d,%f,%s,%s)";
+ private static long prevPartitionInterval;
+
+ @BeforeClass
+ public static void setUp() throws Exception {
+ prevPartitionInterval = IoTDBDescriptor.getInstance().getConfig().getPartitionInterval();
+ ConfigFactory.getConfig().setPartitionInterval(1000);
+ EnvFactory.getEnv().initBeforeClass();
+ prepareData();
+ }
+
+ @AfterClass
+ public static void tearDown() throws Exception {
+ EnvFactory.getEnv().cleanAfterClass();
+ ConfigFactory.getConfig().setPartitionInterval(prevPartitionInterval);
+ }
+
+ // Test FunctionExpression, NegativeExpression and BinaryExpression with time filter
+ @Test
+ public void complexExpressionsWithTimeFilterTest() {
+ Object[] retResults = {
+ 0L, 2L, 6.950000047683716D, 7.0D, 0.9092974268256817D, 1.9092974268256817D, -2L, 2L
+ };
+ try (Connection connection = EnvFactory.getEnv().getConnection();
+ Statement statement = connection.createStatement()) {
+ String query =
+ "SELECT"
+ + " count(temperature)"
+ + ", count(temperature) + avg(temperature) "
+ + ", count(temperature) + max_time(hardware)"
+ + ", sin(count(temperature))"
+ + ", sin(count(temperature)) + 1"
+ + ", -count(temperature)"
+ + ", count(temperature)"
+ + " FROM root.ln.wf01.wt01 WHERE time > 3";
+ boolean hasResultSet = statement.execute(query);
+
+ Assert.assertTrue(hasResultSet);
+ try (ResultSet resultSet = statement.getResultSet()) {
+ Assert.assertTrue(resultSet.next());
+ Assert.assertEquals((long) retResults[0], resultSet.getLong(TIMESTAMP_STR));
+ Assert.assertEquals((long) retResults[1], resultSet.getLong(1));
+ Assert.assertEquals((double) retResults[2], resultSet.getDouble(2), E);
+ Assert.assertEquals((double) retResults[3], resultSet.getDouble(3), E);
+ Assert.assertEquals((double) retResults[4], resultSet.getDouble(4), E);
+ Assert.assertEquals((double) retResults[5], resultSet.getDouble(5), E);
+ Assert.assertEquals((long) retResults[6], resultSet.getLong(6), E);
+ Assert.assertEquals((long) retResults[7], resultSet.getLong(7), E);
+ Assert.assertFalse(resultSet.next());
+ }
+
+ hasResultSet = statement.execute(query + " ORDER BY TIME DESC");
+
+ Assert.assertTrue(hasResultSet);
+ try (ResultSet resultSet = statement.getResultSet()) {
+ Assert.assertTrue(resultSet.next());
+ Assert.assertEquals((long) retResults[0], resultSet.getLong(TIMESTAMP_STR));
+ Assert.assertEquals((long) retResults[1], resultSet.getLong(1));
+ Assert.assertEquals((double) retResults[2], resultSet.getDouble(2), E);
+ Assert.assertEquals((double) retResults[3], resultSet.getDouble(3), E);
+ Assert.assertEquals((double) retResults[4], resultSet.getDouble(4), E);
+ Assert.assertEquals((double) retResults[5], resultSet.getDouble(5), E);
+ Assert.assertEquals((long) retResults[6], resultSet.getLong(6), E);
+ Assert.assertEquals((long) retResults[7], resultSet.getLong(7), E);
+ Assert.assertFalse(resultSet.next());
+ }
+ } catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ @Test
+ public void complexExpressionsWithTimeFilterTest2() {
+ Object[] retResults = {0L, 1.14112000806D, 0.28224001612D};
+ try (Connection connection = EnvFactory.getEnv().getConnection();
+ Statement statement = connection.createStatement()) {
+ String query =
+ "SELECT sin(count(temperature) + 1) + 1,sin(count(temperature) + 1) * 2 FROM root.ln.wf01.wt01 WHERE time > 3";
+ boolean hasResultSet = statement.execute(query);
+
+ Assert.assertTrue(hasResultSet);
+ try (ResultSet resultSet = statement.getResultSet()) {
+ Assert.assertTrue(resultSet.next());
+ Assert.assertEquals((long) retResults[0], resultSet.getLong(TIMESTAMP_STR));
+ Assert.assertEquals((double) retResults[1], resultSet.getDouble(1), E);
+ Assert.assertEquals((double) retResults[2], resultSet.getDouble(2), E);
+ Assert.assertFalse(resultSet.next());
+ }
+
+ hasResultSet = statement.execute(query + " ORDER BY TIME DESC");
+
+ Assert.assertTrue(hasResultSet);
+ try (ResultSet resultSet = statement.getResultSet()) {
+ Assert.assertTrue(resultSet.next());
+ Assert.assertEquals((long) retResults[0], resultSet.getLong(TIMESTAMP_STR));
+ Assert.assertEquals((double) retResults[1], resultSet.getDouble(1), E);
+ Assert.assertEquals((double) retResults[2], resultSet.getDouble(2), E);
+ Assert.assertFalse(resultSet.next());
+ }
+ } catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ // Test FunctionExpression, NegativeExpression and BinaryExpression with value filter
+ @Test
+ public void complexExpressionsWithValueFilterTest() {
+ Object[] retResults = {0L, 4L, 6.75D, 8.0D, -0.7568024953D, 5.0D, -4L, 4L};
+ try (Connection connection = EnvFactory.getEnv().getConnection();
+ Statement statement = connection.createStatement()) {
+ String query =
+ "SELECT"
+ + " count(temperature)"
+ + ", count(temperature) + avg(temperature) "
+ + ", count(temperature) + max_time(hardware)"
+ + ", sin(count(temperature))"
+ + ", count(temperature) + 1"
+ + ", -count(temperature)"
+ + ", count(temperature)"
+ + " FROM root.ln.wf01.wt01 WHERE temperature < 5";
+ boolean hasResultSet = statement.execute(query);
+
+ Assert.assertTrue(hasResultSet);
+ try (ResultSet resultSet = statement.getResultSet()) {
+ Assert.assertTrue(resultSet.next());
+ Assert.assertEquals((long) retResults[0], resultSet.getLong(TIMESTAMP_STR));
+ Assert.assertEquals((long) retResults[1], resultSet.getLong(1));
+ Assert.assertEquals((double) retResults[2], resultSet.getDouble(2), E);
+ Assert.assertEquals((double) retResults[3], resultSet.getDouble(3), E);
+ Assert.assertEquals((double) retResults[4], resultSet.getDouble(4), E);
+ Assert.assertEquals((double) retResults[5], resultSet.getDouble(5), E);
+ Assert.assertEquals((long) retResults[6], resultSet.getLong(6), E);
+ Assert.assertEquals((long) retResults[7], resultSet.getLong(7), E);
+ Assert.assertFalse(resultSet.next());
+ }
+
+ hasResultSet = statement.execute(query + " ORDER BY TIME DESC");
+
+ Assert.assertTrue(hasResultSet);
+ try (ResultSet resultSet = statement.getResultSet()) {
+ Assert.assertTrue(resultSet.next());
+ Assert.assertEquals((long) retResults[0], resultSet.getLong(TIMESTAMP_STR));
+ Assert.assertEquals((long) retResults[1], resultSet.getLong(1));
+ Assert.assertEquals((double) retResults[2], resultSet.getDouble(2), E);
+ Assert.assertEquals((double) retResults[3], resultSet.getDouble(3), E);
+ Assert.assertEquals((double) retResults[4], resultSet.getDouble(4), E);
+ Assert.assertEquals((double) retResults[5], resultSet.getDouble(5), E);
+ Assert.assertEquals((long) retResults[6], resultSet.getLong(6), E);
+ Assert.assertEquals((long) retResults[7], resultSet.getLong(7), E);
+ Assert.assertFalse(resultSet.next());
+ }
+ } catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ // Test FunctionExpression, NegativeExpression and BinaryExpression with wildcard
+ @Test
+ public void complexExpressionWithWildcardTest() throws Exception {
+ double[] retArray =
+ new double[] {
+ 91.83,
+ 151.58,
+ 92.83,
+ 152.58,
+ -0.66224654127,
+ 0.70580058691,
+ -91.83,
+ -151.58,
+ 91.83,
+ 151.58,
+ };
+ try (Connection connection = EnvFactory.getEnv().getConnection();
+ Statement statement = connection.createStatement()) {
+ statement.execute(
+ "select sum(temperature), sum(temperature) + 1, sin(sum(temperature)), -sum(temperature), sum(temperature) from root.sg1.**");
+
+ try (ResultSet resultSet = statement.getResultSet()) {
+ Assert.assertTrue(resultSet.next());
+ Assert.assertEquals(0L, resultSet.getLong(TIMESTAMP_STR));
+ for (int i = 1; i <= 10; i++) {
+ Assert.assertEquals(
+ resultSet.getMetaData().getColumnName(i), retArray[i - 1], resultSet.getDouble(i), E);
+ }
+ Assert.assertFalse(resultSet.next());
+ }
+ }
+ }
+
+ @Test
+ public void complexExpressionWithGroupByTimeWithTimeFilterTest() {
+ Object[][] retArray = {
+ {0L, null, null, null, null, null, null},
+ {1000L, null, null, null, null, null, null},
+ {2000L, null, null, null, null, null, null},
+ {3000L, null, null, null, null, null, null},
+ {4000L, null, null, null, null, null, null},
+ {5000L, null, null, null, null, null, null},
+ {6000L, -0.32466497009092804D, 6499500.0D, 1.64995E7D, 8254.75D, -6499500.0D, 6499500.0D},
+ {7000L, -0.6648173658103295D, 3874750.0D, 1.387475E7D, 10129.75D, -3874750.0D, 3874750.0D},
+ {8000L, 0.8033729538397475D, 4124750.0D, 1.412475E7D, 10879.75D, -4124750.0D, 4124750.0D},
+ };
+
+ try (Connection connection = EnvFactory.getEnv().getConnection();
+ Statement statement = connection.createStatement()) {
+ String query =
+ "SELECT sin(sum(s0)), sum(s0), sum(s0)+10000000, ((avg(s2)-1000) * 3 + 11) / 2, -sum(s0), sum(s0) "
+ + "FROM root.vehicle.d0 WHERE time >= 6000 AND time <= 9000 GROUP BY([0, 9000), 1s)";
+ boolean hasResultSet = statement.execute(query);
+
+ Assert.assertTrue(hasResultSet);
+ int cnt = 0;
+ try (ResultSet resultSet = statement.getResultSet()) {
+ while (resultSet.next()) {
+ Assert.assertEquals((long) retArray[cnt][0], resultSet.getLong(TIMESTAMP_STR));
+ for (int i = 2; i <= 7; i++) {
+ if (retArray[cnt][1] == null) {
+ Assert.assertNull(resultSet.getMetaData().getColumnName(i), resultSet.getObject(i));
+ } else {
+ Assert.assertEquals(
+ resultSet.getMetaData().getColumnName(i),
+ (double) retArray[cnt][i - 1],
+ resultSet.getDouble(i),
+ E);
+ }
+ }
+ cnt++;
+ }
+ }
+ Assert.assertEquals(retArray.length, cnt);
+ // keep the correctness of `order by time desc`
+ hasResultSet = statement.execute(query + " ORDER BY TIME DESC");
+
+ Assert.assertTrue(hasResultSet);
+ cnt = retArray.length;
+ try (ResultSet resultSet = statement.getResultSet()) {
+ while (resultSet.next()) {
+ cnt--;
+ Assert.assertEquals((long) retArray[cnt][0], resultSet.getLong(TIMESTAMP_STR));
+ for (int i = 2; i <= 7; i++) {
+ if (retArray[cnt][1] == null) {
+ Assert.assertNull(resultSet.getMetaData().getColumnName(i), resultSet.getObject(i));
+ } else {
+ Assert.assertEquals(
+ resultSet.getMetaData().getColumnName(i),
+ (double) retArray[cnt][i - 1],
+ resultSet.getDouble(i),
+ E);
+ }
+ }
+ }
+ }
+ Assert.assertEquals(0, cnt);
+
+ } catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ @Test
+ public void complexExpressionWithGroupByTimeWithValueFilterTest() {
+ Object[][] retArray = {
+ {0L, null, null, null, null, null, null},
+ {1000L, null, null, null, null, null, null},
+ {2000L, null, null, null, null, null, null},
+ {3000L, null, null, null, null, null, null},
+ {4000L, null, null, null, null, null, null},
+ {5000L, null, null, null, null, null, null},
+ {6000L, -0.32466497009092804D, 6499500.0D, 1.64995E7D, 8254.75D, -6499500.0D, 6499500.0D},
+ {7000L, -0.6648173658103295D, 3874750.0D, 1.387475E7D, 10129.75D, -3874750.0D, 3874750.0D},
+ {8000L, 0.8033729538397475D, 4124750.0D, 1.412475E7D, 10879.75D, -4124750.0D, 4124750.0D},
+ };
+
+ try (Connection connection = EnvFactory.getEnv().getConnection();
+ Statement statement = connection.createStatement()) {
+ String query =
+ "SELECT sin(sum(s0)), sum(s0), sum(s0)+10000000, ((avg(s2)-1000) * 3 + 11) / 2, -sum(s0), sum(s0) "
+ + "FROM root.vehicle.d0 WHERE s0 >= 6000 AND s0 <= 9000 GROUP BY([0, 9000), 1s)";
+ boolean hasResultSet = statement.execute(query);
+
+ Assert.assertTrue(hasResultSet);
+ int cnt = 0;
+ try (ResultSet resultSet = statement.getResultSet()) {
+ while (resultSet.next()) {
+ Assert.assertEquals((long) retArray[cnt][0], resultSet.getLong(TIMESTAMP_STR));
+ for (int i = 2; i <= 7; i++) {
+ if (retArray[cnt][1] == null) {
+ Assert.assertNull(resultSet.getMetaData().getColumnName(i), resultSet.getObject(i));
+ } else {
+ Assert.assertEquals(
+ resultSet.getMetaData().getColumnName(i),
+ (double) retArray[cnt][i - 1],
+ resultSet.getDouble(i),
+ E);
+ }
+ }
+ cnt++;
+ }
+ }
+ Assert.assertEquals(retArray.length, cnt);
+ // keep the correctness of `order by time desc`
+ hasResultSet = statement.execute(query + " ORDER BY TIME DESC");
+
+ Assert.assertTrue(hasResultSet);
+ cnt = retArray.length;
+ try (ResultSet resultSet = statement.getResultSet()) {
+ while (resultSet.next()) {
+ cnt--;
+ Assert.assertEquals((long) retArray[cnt][0], resultSet.getLong(TIMESTAMP_STR));
+ for (int i = 2; i <= 7; i++) {
+ if (retArray[cnt][1] == null) {
+ Assert.assertNull(resultSet.getMetaData().getColumnName(i), resultSet.getObject(i));
+ } else {
+ Assert.assertEquals(
+ resultSet.getMetaData().getColumnName(i),
+ (double) retArray[cnt][i - 1],
+ resultSet.getDouble(i),
+ E);
+ }
+ }
+ }
+ }
+ Assert.assertEquals(0, cnt);
+
+ } catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ private static void prepareData() {
+ try (Connection connection = EnvFactory.getEnv().getConnection();
+ Statement statement = connection.createStatement()) {
+
+ for (String sql : creationSqls) {
+ statement.execute(sql);
+ }
+
+ // prepare BufferWrite file
+ for (int i = 5000; i < 7000; i++) {
+ statement.execute(
+ String.format(
+ Locale.ENGLISH, insertTemplate, i, i, i, (double) i, "'" + i + "'", "true"));
+ }
+ statement.execute("FLUSH");
+ for (int i = 7500; i < 8500; i++) {
+ statement.execute(
+ String.format(
+ Locale.ENGLISH, insertTemplate, i, i, i, (double) i, "'" + i + "'", "false"));
+ }
+ statement.execute("FLUSH");
+ // prepare Unseq-File
+ for (int i = 500; i < 1500; i++) {
+ statement.execute(
+ String.format(
+ Locale.ENGLISH, insertTemplate, i, i, i, (double) i, "'" + i + "'", "true"));
+ }
+ statement.execute("FLUSH");
+ for (int i = 3000; i < 6500; i++) {
+ statement.execute(
+ String.format(
+ Locale.ENGLISH, insertTemplate, i, i, i, (double) i, "'" + i + "'", "false"));
+ }
+ statement.execute("merge");
+
+ // prepare BufferWrite cache
+ for (int i = 9000; i < 10000; i++) {
+ statement.execute(
+ String.format(
+ Locale.ENGLISH, insertTemplate, i, i, i, (double) i, "'" + i + "'", "true"));
+ }
+ // prepare Overflow cache
+ for (int i = 2000; i < 2500; i++) {
+ statement.execute(
+ String.format(
+ Locale.ENGLISH, insertTemplate, i, i, i, (double) i, "'" + i + "'", "false"));
+ }
+
+ for (String sql : dataSet3) {
+ statement.execute(sql);
+ }
+
+ for (String sql : dataSet2) {
+ statement.execute(sql);
+ }
+
+ for (String sql : dataSet4) {
+ statement.execute(sql);
+ }
+
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+ }
+}
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/executor/PlanExecutor.java b/server/src/main/java/org/apache/iotdb/db/qp/executor/PlanExecutor.java
index a7052fa..935903c 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/executor/PlanExecutor.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/executor/PlanExecutor.java
@@ -80,6 +80,7 @@ import org.apache.iotdb.db.qp.physical.crud.LastQueryPlan;
import org.apache.iotdb.db.qp.physical.crud.QueryIndexPlan;
import org.apache.iotdb.db.qp.physical.crud.QueryPlan;
import org.apache.iotdb.db.qp.physical.crud.RawDataQueryPlan;
+import org.apache.iotdb.db.qp.physical.crud.UDAFPlan;
import org.apache.iotdb.db.qp.physical.crud.UDTFPlan;
import org.apache.iotdb.db.qp.physical.sys.ActivateTemplatePlan;
import org.apache.iotdb.db.qp.physical.sys.AlterTimeSeriesPlan;
@@ -588,6 +589,9 @@ public class PlanExecutor implements IPlanExecutor {
if (queryPlan.getPaths() == null || queryPlan.getPaths().isEmpty()) {
// no time series are selected, return EmptyDataSet
return new EmptyDataSet();
+ } else if (queryPlan instanceof UDAFPlan) {
+ UDAFPlan udafPlan = (UDAFPlan) queryPlan;
+ queryDataSet = queryRouter.udafQuery(udafPlan, context);
} else if (queryPlan instanceof UDTFPlan) {
UDTFPlan udtfPlan = (UDTFPlan) queryPlan;
queryDataSet = queryRouter.udtfQuery(udtfPlan, context);
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/logical/crud/AggregationQueryOperator.java b/server/src/main/java/org/apache/iotdb/db/qp/logical/crud/AggregationQueryOperator.java
index 1b2e865..36719c7 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/logical/crud/AggregationQueryOperator.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/logical/crud/AggregationQueryOperator.java
@@ -58,7 +58,14 @@ public class AggregationQueryOperator extends QueryOperator {
if (!isAlignByTime()) {
throw new LogicalOperatorException("AGGREGATION doesn't support disable align clause.");
}
+ checkSelectComponent(selectComponent);
+ if (isGroupByLevel() && isAlignByDevice()) {
+ throw new LogicalOperatorException("group by level does not support align by device now.");
+ }
+ }
+ protected void checkSelectComponent(SelectComponent selectComponent)
+ throws LogicalOperatorException {
if (hasTimeSeriesGeneratingFunction()) {
throw new LogicalOperatorException(
"User-defined and built-in hybrid aggregation is not supported together.");
@@ -71,17 +78,12 @@ public class AggregationQueryOperator extends QueryOperator {
}
// Currently, the aggregation function expression can only contain a timeseries operand.
if (expression instanceof FunctionExpression
- && (((FunctionExpression) expression).getExpressions().size() != 1
- || !(((FunctionExpression) expression).getExpressions().get(0)
- instanceof TimeSeriesOperand))) {
+ && (expression.getExpressions().size() != 1
+ || !(expression.getExpressions().get(0) instanceof TimeSeriesOperand))) {
throw new LogicalOperatorException(
"The argument of the aggregation function must be a time series.");
}
}
-
- if (isGroupByLevel() && isAlignByDevice()) {
- throw new LogicalOperatorException("group by level does not support align by device now.");
- }
}
@Override
@@ -125,16 +127,20 @@ public class AggregationQueryOperator extends QueryOperator {
AggregationPlan aggregationPlan = (AggregationPlan) queryPlan;
aggregationPlan.setAggregations(selectComponent.getAggregationFunctions());
if (isGroupByLevel()) {
- aggregationPlan.setLevels(specialClauseComponent.getLevels());
- aggregationPlan.setGroupByLevelController(specialClauseComponent.groupByLevelController);
- try {
- if (!verifyAllAggregationDataTypesEqual()) {
- throw new LogicalOperatorException("Aggregate among unmatched data types");
- }
- } catch (MetadataException e) {
- throw new LogicalOperatorException(e);
- }
+ initGroupByLevel(aggregationPlan);
}
return aggregationPlan;
}
+
+ protected void initGroupByLevel(AggregationPlan aggregationPlan) throws QueryProcessException {
+ aggregationPlan.setLevels(specialClauseComponent.getLevels());
+ aggregationPlan.setGroupByLevelController(specialClauseComponent.groupByLevelController);
+ try {
+ if (!verifyAllAggregationDataTypesEqual()) {
+ throw new LogicalOperatorException("Aggregate among unmatched data types");
+ }
+ } catch (MetadataException e) {
+ throw new LogicalOperatorException(e);
+ }
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/logical/crud/FillQueryOperator.java b/server/src/main/java/org/apache/iotdb/db/qp/logical/crud/FillQueryOperator.java
index a285424..3f05db9 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/logical/crud/FillQueryOperator.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/logical/crud/FillQueryOperator.java
@@ -41,7 +41,7 @@ public class FillQueryOperator extends QueryOperator {
throw new LogicalOperatorException("FILL doesn't support disable align clause.");
}
- if (hasTimeSeriesGeneratingFunction()) {
+ if (hasTimeSeriesGeneratingFunction() || selectComponent.hasUserDefinedAggregationFunction()) {
throw new LogicalOperatorException("Fill functions are not supported in UDF queries.");
}
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/logical/crud/QueryOperator.java b/server/src/main/java/org/apache/iotdb/db/qp/logical/crud/QueryOperator.java
index 0d2fb81..0082c20 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/logical/crud/QueryOperator.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/logical/crud/QueryOperator.java
@@ -196,17 +196,7 @@ public class QueryOperator extends Operator {
// transform filter operator to expression
if (whereComponent != null) {
- FilterOperator filterOperator = whereComponent.getFilterOperator();
- List<PartialPath> filterPaths = new ArrayList<>(filterOperator.getPathSet());
- HashMap<PartialPath, TSDataType> pathTSDataTypeHashMap = new HashMap<>();
- for (PartialPath filterPath : filterPaths) {
- rawDataQueryPlan.addFilterPathInDeviceToMeasurements(filterPath);
- pathTSDataTypeHashMap.put(
- filterPath,
- SQLConstant.isReservedPath(filterPath) ? TSDataType.INT64 : filterPath.getSeriesType());
- }
- IExpression expression = filterOperator.transformToExpression(pathTSDataTypeHashMap);
- rawDataQueryPlan.setExpression(expression);
+ transformFilterOperatorToExpression(generator, rawDataQueryPlan);
}
if (queryPlan instanceof QueryIndexPlan) {
@@ -226,6 +216,21 @@ public class QueryOperator extends Operator {
return rawDataQueryPlan;
}
+ protected void transformFilterOperatorToExpression(
+ PhysicalGenerator generator, RawDataQueryPlan rawDataQueryPlan) throws QueryProcessException {
+ FilterOperator filterOperator = whereComponent.getFilterOperator();
+ List<PartialPath> filterPaths = new ArrayList<>(filterOperator.getPathSet());
+ HashMap<PartialPath, TSDataType> pathTSDataTypeHashMap = new HashMap<>();
+ for (PartialPath filterPath : filterPaths) {
+ rawDataQueryPlan.addFilterPathInDeviceToMeasurements(filterPath);
+ pathTSDataTypeHashMap.put(
+ filterPath,
+ SQLConstant.isReservedPath(filterPath) ? TSDataType.INT64 : filterPath.getSeriesType());
+ }
+ IExpression expression = filterOperator.transformToExpression(pathTSDataTypeHashMap);
+ rawDataQueryPlan.setExpression(expression);
+ }
+
protected AlignByDevicePlan generateAlignByDevicePlan(PhysicalGenerator generator)
throws QueryProcessException {
AlignByDevicePlan alignByDevicePlan = new AlignByDevicePlan();
@@ -329,7 +334,7 @@ public class QueryOperator extends Operator {
return alignByDevicePlan;
}
- private void convertSpecialClauseValues(QueryPlan queryPlan) {
+ protected void convertSpecialClauseValues(QueryPlan queryPlan) {
if (specialClauseComponent != null) {
queryPlan.setWithoutAllNull(specialClauseComponent.isWithoutAllNull());
queryPlan.setWithoutAnyNull(specialClauseComponent.isWithoutAnyNull());
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/logical/crud/SelectComponent.java b/server/src/main/java/org/apache/iotdb/db/qp/logical/crud/SelectComponent.java
index 141d0ff..aed3727 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/logical/crud/SelectComponent.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/logical/crud/SelectComponent.java
@@ -36,6 +36,7 @@ public final class SelectComponent {
private boolean hasAggregationFunction = false;
private boolean hasTimeSeriesGeneratingFunction = false;
+ private boolean hasUserDefinedAggregationFunction = false;
private List<ResultColumn> resultColumns = new ArrayList<>();
@@ -66,8 +67,16 @@ public final class SelectComponent {
return hasTimeSeriesGeneratingFunction;
}
+ public boolean hasUserDefinedAggregationFunction() {
+ return hasUserDefinedAggregationFunction;
+ }
+
public void addResultColumn(ResultColumn resultColumn) {
resultColumns.add(resultColumn);
+ if (resultColumn.getExpression().isUDAFExpression()) {
+ hasUserDefinedAggregationFunction = true;
+ }
+
if (resultColumn.getExpression().isAggregationFunctionExpression()) {
hasAggregationFunction = true;
}
@@ -108,9 +117,6 @@ public final class SelectComponent {
}
public List<String> getAggregationFunctions() {
- if (!hasAggregationFunction()) {
- return null;
- }
if (aggregationFunctionsCache == null) {
aggregationFunctionsCache = new ArrayList<>();
for (ResultColumn resultColumn : resultColumns) {
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/logical/crud/UDAFQueryOperator.java b/server/src/main/java/org/apache/iotdb/db/qp/logical/crud/UDAFQueryOperator.java
new file mode 100644
index 0000000..d6a14bd
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/qp/logical/crud/UDAFQueryOperator.java
@@ -0,0 +1,172 @@
+/*
+ * 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.qp.logical.crud;
+
+import org.apache.iotdb.db.exception.query.LogicalOperatorException;
+import org.apache.iotdb.db.exception.query.QueryProcessException;
+import org.apache.iotdb.db.qp.physical.PhysicalPlan;
+import org.apache.iotdb.db.qp.physical.crud.AggregationPlan;
+import org.apache.iotdb.db.qp.physical.crud.UDAFPlan;
+import org.apache.iotdb.db.qp.strategy.PhysicalGenerator;
+import org.apache.iotdb.db.query.expression.Expression;
+import org.apache.iotdb.db.query.expression.ResultColumn;
+import org.apache.iotdb.db.query.expression.unary.FunctionExpression;
+import org.apache.iotdb.db.query.expression.unary.TimeSeriesOperand;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.Map;
+
+/**
+ * For a UDAFPlan, we construct an inner AggregationPlan for it. Example: select
+ * count(a)/count(b),count(a)+sum(b) from root.sg To init inner AggregationPlan, we will convert it
+ * to statement: select count(a),count(b),count(a),sum(b) from root.sg. innerResultColumnsCache will
+ * be [count(a),count(b),sum(b)].
+ */
+public class UDAFQueryOperator extends QueryOperator {
+
+ private ArrayList<ResultColumn> innerResultColumnsCache;
+
+ private AggregationQueryOperator aggrOp;
+
+ public UDAFQueryOperator(AggregationQueryOperator queryOperator) {
+ super(queryOperator);
+ this.aggrOp = queryOperator;
+ }
+
+ @Override
+ public void check() throws LogicalOperatorException {
+ super.check();
+
+ if (!isAlignByTime()) {
+ throw new LogicalOperatorException("AGGREGATION doesn't support disable align clause.");
+ }
+ checkSelectComponent(selectComponent);
+ if (isGroupByLevel()) {
+ throw new LogicalOperatorException(
+ "UDF nesting aggregations in GROUP BY query does not support grouping by level now.");
+ }
+ if (aggrOp instanceof GroupByFillQueryOperator) {
+ throw new LogicalOperatorException(
+ "UDF nesting aggregations in GROUP BY query does not support FILL now.");
+ }
+ }
+
+ private void checkSelectComponent(SelectComponent selectComponent)
+ throws LogicalOperatorException {
+ for (ResultColumn resultColumn : selectComponent.getResultColumns()) {
+ Expression expression = resultColumn.getExpression();
+ checkEachExpression(expression);
+ }
+ }
+
+ public ArrayList<ResultColumn> getInnerResultColumnsCache() {
+ if (innerResultColumnsCache == null) {
+ innerResultColumnsCache = new ArrayList<>();
+ for (ResultColumn resultColumn : selectComponent.getResultColumns()) {
+ Expression expression = resultColumn.getExpression();
+ addInnerResultColumn(expression);
+ }
+ }
+ return innerResultColumnsCache;
+ }
+
+ private void addInnerResultColumn(Expression expression) {
+ for (Iterator<Expression> it = expression.iterator(); it.hasNext(); ) {
+ Expression currentExp = it.next();
+ if (currentExp.isAggregationFunctionExpression()) {
+ innerResultColumnsCache.add(new ResultColumn(currentExp));
+ }
+ }
+ }
+
+ @Override
+ public PhysicalPlan generatePhysicalPlan(PhysicalGenerator generator)
+ throws QueryProcessException {
+ SelectComponent copiedSelectComponent = new SelectComponent(getSelectComponent());
+ copiedSelectComponent.setResultColumns(getInnerResultColumnsCache());
+ aggrOp.setSelectComponent(copiedSelectComponent);
+ aggrOp.setFromComponent(getFromComponent());
+ aggrOp.setWhereComponent(getWhereComponent());
+ aggrOp.setSpecialClauseComponent(getSpecialClauseComponent());
+ aggrOp.setProps(getProps());
+ aggrOp.setIndexType(getIndexType());
+ aggrOp.setEnableTracing(isEnableTracing());
+ AggregationPlan innerAggregationPlan = (AggregationPlan) aggrOp.generatePhysicalPlan(generator);
+ PhysicalPlan physicalPlan;
+ if (!isAlignByDevice()) {
+ physicalPlan =
+ super.generateRawDataQueryPlan(generator, new UDAFPlan(selectComponent.getZoneId()));
+ UDAFPlan udafPlan = (UDAFPlan) physicalPlan;
+ udafPlan.setInnerAggregationPlan(innerAggregationPlan);
+ Map<Expression, Integer> expressionToInnerResultIndexMap = new HashMap<>();
+ // The following codes are used to establish a link from AggregationResult to expression tree
+ // input.
+ // For example:
+ // Expressions [ avg(s1) + 1, avg(s1) + avg(s2), sin(avg(s2)) ]
+ // | | | |
+ // Inner result columns [ avg(s1) , avg(s1) , avg(s2) avg(s2) ]
+ // \ / \ /
+ // deduplicated [ avg(s1), avg(s2) ]
+ // \ /
+ // expressionToInnerResultIndexMap {avg(s1): 0, avg(s2): 1}
+ Map<String, Integer> aggrIndexMap = new HashMap<>();
+ for (int i = 0; i < innerAggregationPlan.getDeduplicatedPaths().size(); i++) {
+ aggrIndexMap.put(
+ innerAggregationPlan.getDeduplicatedAggregations().get(i)
+ + "("
+ + innerAggregationPlan.getDeduplicatedPaths().get(i)
+ + ")",
+ i);
+ }
+ for (ResultColumn rc : getInnerResultColumnsCache()) {
+ expressionToInnerResultIndexMap.put(
+ rc.getExpression(),
+ aggrIndexMap.get(
+ ((FunctionExpression) rc.getExpression()).getExpressionStringInternal()));
+ }
+ udafPlan.setExpressionToInnerResultIndexMap(expressionToInnerResultIndexMap);
+ udafPlan.constructUdfExecutors(selectComponent.getResultColumns());
+ } else {
+ // todo: align by device
+ physicalPlan = new AggregationPlan();
+ }
+ return physicalPlan;
+ }
+
+ private void checkEachExpression(Expression expression) throws LogicalOperatorException {
+ if (expression instanceof TimeSeriesOperand) {
+ throw new LogicalOperatorException(AggregationQueryOperator.ERROR_MESSAGE1);
+ }
+ // Currently, the aggregation function expression can only contain a timeseries operand.
+ if (expression.isAggregationFunctionExpression()) {
+ if (expression.getExpressions().size() == 1
+ && expression.getExpressions().get(0) instanceof TimeSeriesOperand) {
+ return;
+ }
+ throw new LogicalOperatorException(
+ "The argument of the aggregation function must be a time series.");
+ }
+
+ for (Expression childExp : expression.getExpressions()) {
+ checkEachExpression(childExp);
+ }
+ }
+}
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/UDAFPlan.java b/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/UDAFPlan.java
new file mode 100644
index 0000000..4f69ea6
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/UDAFPlan.java
@@ -0,0 +1,68 @@
+/*
+ * 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.qp.physical.crud;
+
+import org.apache.iotdb.db.exception.metadata.MetadataException;
+import org.apache.iotdb.db.qp.logical.Operator.OperatorType;
+import org.apache.iotdb.db.qp.strategy.PhysicalGenerator;
+import org.apache.iotdb.db.query.expression.Expression;
+
+import java.time.ZoneId;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
+/** The physical plan of aggregation query with UDF nested */
+public class UDAFPlan extends UDTFPlan {
+
+ // Construct an innerAggregationPlan using resultColumns of UDAFPlan
+ private AggregationPlan innerAggregationPlan;
+
+ public UDAFPlan(ZoneId zoneId) {
+ super(zoneId);
+ setOperatorType(OperatorType.UDAF);
+ }
+
+ public void setExpressionToInnerResultIndexMap(
+ Map<Expression, Integer> expressionToInnerResultIndexMap) {
+ expressionToInnerResultIndexMap.forEach((k, v) -> pathNameToReaderIndex.put(k.toString(), v));
+ }
+
+ public void setInnerAggregationPlan(AggregationPlan innerAggregationPlan) {
+ this.innerAggregationPlan = innerAggregationPlan;
+ }
+
+ public AggregationPlan getInnerAggregationPlan() {
+ return innerAggregationPlan;
+ }
+
+ @Override
+ public void deduplicate(PhysicalGenerator physicalGenerator) throws MetadataException {
+ Set<String> columnForDisplaySet = new HashSet<>();
+ for (int i = 0; i < resultColumns.size(); i++) {
+ String columnForDisplay = resultColumns.get(i).getResultColumnName();
+ if (!columnForDisplaySet.contains(columnForDisplay)) {
+ int datasetOutputIndex = getPathToIndex().size();
+ setColumnNameToDatasetOutputIndex(columnForDisplay, datasetOutputIndex);
+ setDatasetOutputIndexToResultColumnIndex(datasetOutputIndex, i);
+ columnForDisplaySet.add(columnForDisplay);
+ }
+ }
+ }
+}
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/UDTFPlan.java b/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/UDTFPlan.java
index 28832f4..c88fe2e 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/UDTFPlan.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/UDTFPlan.java
@@ -87,7 +87,7 @@ public class UDTFPlan extends RawDataQueryPlan implements UDFPlan {
}
}
- private void setDatasetOutputIndexToResultColumnIndex(
+ protected void setDatasetOutputIndexToResultColumnIndex(
int datasetOutputIndex, Integer originalIndex) {
datasetOutputIndexToResultColumnIndex.put(datasetOutputIndex, originalIndex);
}
@@ -130,4 +130,8 @@ public class UDTFPlan extends RawDataQueryPlan implements UDFPlan {
public int getReaderIndex(String pathName) {
return pathNameToReaderIndex.get(pathName);
}
+
+ public int getReaderIndexByExpressionName(String expressionName) {
+ return pathNameToReaderIndex.get(expressionName);
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/sql/IoTDBSqlVisitor.java b/server/src/main/java/org/apache/iotdb/db/qp/sql/IoTDBSqlVisitor.java
index 0aeb2d0..2b069c4 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/sql/IoTDBSqlVisitor.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/sql/IoTDBSqlVisitor.java
@@ -49,6 +49,7 @@ import org.apache.iotdb.db.qp.logical.crud.RegexpOperator;
import org.apache.iotdb.db.qp.logical.crud.SelectComponent;
import org.apache.iotdb.db.qp.logical.crud.SelectIntoOperator;
import org.apache.iotdb.db.qp.logical.crud.SpecialClauseComponent;
+import org.apache.iotdb.db.qp.logical.crud.UDAFQueryOperator;
import org.apache.iotdb.db.qp.logical.crud.UDFQueryOperator;
import org.apache.iotdb.db.qp.logical.crud.WhereComponent;
import org.apache.iotdb.db.qp.logical.sys.ActivateTemplateOperator;
@@ -131,7 +132,15 @@ import org.antlr.v4.runtime.tree.TerminalNode;
import java.io.File;
import java.time.ZoneId;
-import java.util.*;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.EnumMap;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
@@ -2453,17 +2462,20 @@ public class IoTDBSqlVisitor extends IoTDBSqlParserBaseVisitor<Operator> {
queryOp = new LastQueryOperator(queryOp);
}
- boolean isFirstElement = true;
for (IoTDBSqlParser.ResultColumnContext resultColumnContext : ctx.resultColumn()) {
selectComponent.addResultColumn(parseResultColumn(resultColumnContext));
- // judge query type according to the first select element
- if (!hasDecidedQueryType()) {
- if (selectComponent.hasAggregationFunction()) {
- queryOp = new AggregationQueryOperator(queryOp);
- } else if (selectComponent.hasTimeSeriesGeneratingFunction()) {
- queryOp = new UDFQueryOperator(queryOp);
- }
+ }
+
+ if (!hasDecidedQueryType()) {
+ if (selectComponent.hasUserDefinedAggregationFunction()) {
+ queryOp = new UDAFQueryOperator(new AggregationQueryOperator(queryOp));
+ } else if (selectComponent.hasAggregationFunction()) {
+ queryOp = new AggregationQueryOperator(queryOp);
+ } else if (selectComponent.hasTimeSeriesGeneratingFunction()) {
+ queryOp = new UDFQueryOperator(queryOp);
}
+ } else if (selectComponent.hasUserDefinedAggregationFunction()) {
+ queryOp = new UDAFQueryOperator((AggregationQueryOperator) (queryOp));
}
queryOp.setSelectComponent(selectComponent);
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/utils/GroupByLevelController.java b/server/src/main/java/org/apache/iotdb/db/qp/utils/GroupByLevelController.java
index 7afa82e..48556cd 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/utils/GroupByLevelController.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/utils/GroupByLevelController.java
@@ -67,49 +67,70 @@ public class GroupByLevelController {
public void control(ResultColumn rawColumn, List<ResultColumn> resultColumns)
throws LogicalOptimizeException {
- boolean isCountStar = ((FunctionExpression) rawColumn.getExpression()).isCountStar();
Iterator<ResultColumn> iterator = resultColumns.iterator();
+
+ // As one expression may have many aggregation results in the tree leaf, here we should traverse
+ // all the successor expressions and record the count(*) indices
+ Set<Integer> countWildcardIterIndices = new HashSet<>();
+ int idx = 0;
+ for (Iterator<Expression> it = rawColumn.getExpression().iterator(); it.hasNext(); ) {
+ Expression expression = it.next();
+ if (expression instanceof FunctionExpression
+ && expression.isAggregationFunctionExpression()
+ && ((FunctionExpression) expression).isCountStar()) {
+ countWildcardIterIndices.add(idx);
+ }
+ idx++;
+ }
for (int i = 0; i < prevSize; i++) {
iterator.next();
}
while (iterator.hasNext()) {
ResultColumn resultColumn = iterator.next();
- Expression expression = resultColumn.getExpression();
- if (expression instanceof FunctionExpression
- && expression.isAggregationFunctionExpression()) {
- List<PartialPath> paths = ((FunctionExpression) expression).getPaths();
- String functionName = ((FunctionExpression) expression).getFunctionName();
- String groupedPath =
- generatePartialPathByLevel(isCountStar, paths.get(0).getNodes(), levels);
- String rawPath = String.format("%s(%s)", functionName, paths.get(0).getFullPath());
- String pathWithFunction = String.format("%s(%s)", functionName, groupedPath);
+ Expression rootExpression = resultColumn.getExpression();
+ boolean hasAggregation = false;
+ idx = 0;
+ for (Iterator<Expression> it = rootExpression.iterator(); it.hasNext(); ) {
+ Expression expression = it.next();
+ if (expression instanceof FunctionExpression
+ && expression.isAggregationFunctionExpression()) {
+ hasAggregation = true;
+ List<PartialPath> paths = ((FunctionExpression) expression).getPaths();
+ String functionName = ((FunctionExpression) expression).getFunctionName();
+ boolean isCountStar = countWildcardIterIndices.contains(idx);
+ String groupedPath =
+ generatePartialPathByLevel(isCountStar, paths.get(0).getNodes(), levels);
+ String rawPath = String.format("%s(%s)", functionName, paths.get(0).getFullPath());
+ String pathWithFunction = String.format("%s(%s)", functionName, groupedPath);
- if (seriesLimit == 0 && seriesOffset == 0) {
- groupedPathMap.put(rawPath, pathWithFunction);
- } else {
- if (seriesOffset > 0 && offsetPaths != null) {
- offsetPaths.add(pathWithFunction);
- if (offsetPaths.size() <= seriesOffset) {
- iterator.remove();
- if (offsetPaths.size() == seriesOffset) {
- seriesOffset = 0;
+ if (seriesLimit == 0 && seriesOffset == 0) {
+ groupedPathMap.put(rawPath, pathWithFunction);
+ } else {
+ if (seriesOffset > 0 && offsetPaths != null) {
+ offsetPaths.add(pathWithFunction);
+ if (offsetPaths.size() <= seriesOffset) {
+ iterator.remove();
+ if (offsetPaths.size() == seriesOffset) {
+ seriesOffset = 0;
+ }
+ }
+ } else if (offsetPaths == null || !offsetPaths.contains(pathWithFunction)) {
+ limitPaths.add(pathWithFunction);
+ if (seriesLimit > 0 && limitPaths.size() > seriesLimit) {
+ iterator.remove();
+ limitPaths.remove(pathWithFunction);
+ } else {
+ groupedPathMap.put(rawPath, pathWithFunction);
}
- }
- } else if (offsetPaths == null || !offsetPaths.contains(pathWithFunction)) {
- limitPaths.add(pathWithFunction);
- if (seriesLimit > 0 && limitPaths.size() > seriesLimit) {
- iterator.remove();
- limitPaths.remove(pathWithFunction);
} else {
- groupedPathMap.put(rawPath, pathWithFunction);
+ iterator.remove();
}
- } else {
- iterator.remove();
}
}
- } else {
- throw new LogicalOptimizeException(
- expression.toString() + "can't be used in group by level.");
+ idx++;
+ }
+ if (!hasAggregation) {
+ throw new LogicalOptimizeException(rootExpression + " can't be used in group by level.");
}
}
prevSize = resultColumns.size();
diff --git a/server/src/main/java/org/apache/iotdb/db/query/dataset/UDFInputDataSet.java b/server/src/main/java/org/apache/iotdb/db/query/dataset/IUDFInputDataSet.java
similarity index 77%
copy from server/src/main/java/org/apache/iotdb/db/query/dataset/UDFInputDataSet.java
copy to server/src/main/java/org/apache/iotdb/db/query/dataset/IUDFInputDataSet.java
index d404288..ecb2eb7 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/dataset/UDFInputDataSet.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/dataset/IUDFInputDataSet.java
@@ -24,11 +24,19 @@ import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import java.io.IOException;
import java.util.List;
-public interface UDFInputDataSet {
+/** The input data set interface for a UDFPlan */
+public interface IUDFInputDataSet {
+ /** returns the input data types, except the timestamp column(the last column). */
List<TSDataType> getDataTypes();
+ /** Whether the data set has next row. */
boolean hasNextRowInObjects() throws IOException;
+ /**
+ * Get the next row for UDFPlan input.
+ *
+ * <p>The last element in the row is the timestamp.
+ */
Object[] nextRowInObjects() throws IOException;
}
diff --git a/server/src/main/java/org/apache/iotdb/db/query/dataset/RawQueryDataSetWithValueFilter.java b/server/src/main/java/org/apache/iotdb/db/query/dataset/RawQueryDataSetWithValueFilter.java
index b0d9f5b..630e45b 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/dataset/RawQueryDataSetWithValueFilter.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/dataset/RawQueryDataSetWithValueFilter.java
@@ -30,7 +30,7 @@ import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
-public class RawQueryDataSetWithValueFilter extends QueryDataSet implements UDFInputDataSet {
+public class RawQueryDataSetWithValueFilter extends QueryDataSet implements IUDFInputDataSet {
private final TimeGenerator timeGenerator;
private final List<IReaderByTimestamp> seriesReaderByTimestampList;
diff --git a/server/src/main/java/org/apache/iotdb/db/query/dataset/UDFInputDataSet.java b/server/src/main/java/org/apache/iotdb/db/query/dataset/UDFInputDataSet.java
index d404288..73bb0c7 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/dataset/UDFInputDataSet.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/dataset/UDFInputDataSet.java
@@ -16,19 +16,51 @@
* specific language governing permissions and limitations
* under the License.
*/
-
package org.apache.iotdb.db.query.dataset;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.read.common.Field;
+import org.apache.iotdb.tsfile.read.common.RowRecord;
+import org.apache.iotdb.tsfile.read.query.dataset.QueryDataSet;
import java.io.IOException;
import java.util.List;
-public interface UDFInputDataSet {
+/**
+ * the input data set of an UDAF query. It accepts any query results as input instead of raw
+ * timeseries data
+ */
+public class UDFInputDataSet implements IUDFInputDataSet {
+
+ private final QueryDataSet dataSet;
+ private List<TSDataType> dataTypes;
+
+ public UDFInputDataSet(QueryDataSet dataSet, List<TSDataType> dataTypes) {
+ this.dataSet = dataSet;
+ this.dataTypes = dataTypes;
+ }
- List<TSDataType> getDataTypes();
+ @Override
+ public List<TSDataType> getDataTypes() {
+ return dataTypes;
+ }
- boolean hasNextRowInObjects() throws IOException;
+ @Override
+ public boolean hasNextRowInObjects() throws IOException {
+ return dataSet.hasNextWithoutConstraint();
+ }
- Object[] nextRowInObjects() throws IOException;
+ @Override
+ public Object[] nextRowInObjects() throws IOException {
+ Object[] nextRow = new Object[dataSet.getColumnNum() + 1];
+ RowRecord r = dataSet.nextWithoutConstraint();
+ for (int i = 0; i < dataSet.getColumnNum(); i++) {
+ Field f = r.getFields().get(i);
+ if (f != null) {
+ nextRow[i] = f.getObjectValue(f.getDataType());
+ }
+ }
+ nextRow[dataSet.getColumnNum()] = r.getTimestamp();
+ return nextRow;
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/query/dataset/UDFRawQueryInputDataSetWithoutValueFilter.java b/server/src/main/java/org/apache/iotdb/db/query/dataset/UDFRawQueryInputDataSetWithoutValueFilter.java
index 7cdf680..f4e4cf8 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/dataset/UDFRawQueryInputDataSetWithoutValueFilter.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/dataset/UDFRawQueryInputDataSetWithoutValueFilter.java
@@ -27,7 +27,7 @@ import java.io.IOException;
import java.util.List;
public class UDFRawQueryInputDataSetWithoutValueFilter extends RawQueryDataSetWithoutValueFilter
- implements UDFInputDataSet {
+ implements IUDFInputDataSet {
public UDFRawQueryInputDataSetWithoutValueFilter(
long queryId,
diff --git a/server/src/main/java/org/apache/iotdb/db/query/dataset/UDTFAlignByTimeDataSet.java b/server/src/main/java/org/apache/iotdb/db/query/dataset/UDTFAlignByTimeDataSet.java
index d2cbf09..7a6b37d 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/dataset/UDTFAlignByTimeDataSet.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/dataset/UDTFAlignByTimeDataSet.java
@@ -44,6 +44,7 @@ import java.util.List;
public class UDTFAlignByTimeDataSet extends UDTFDataSet implements DirectAlignByTimeDataSet {
protected TimeSelector timeHeap;
+ private final boolean keepNull;
/** execute with value filter */
public UDTFAlignByTimeDataSet(
@@ -61,6 +62,7 @@ public class UDTFAlignByTimeDataSet extends UDTFDataSet implements DirectAlignBy
timestampGenerator,
readersOfSelectedSeries,
cached);
+ keepNull = false;
initTimeHeap();
}
@@ -74,17 +76,22 @@ public class UDTFAlignByTimeDataSet extends UDTFDataSet implements DirectAlignBy
udtfPlan.getDeduplicatedPaths(),
udtfPlan.getDeduplicatedDataTypes(),
readersOfSelectedSeries);
+ keepNull = false;
+ initTimeHeap();
+ }
+
+ public UDTFAlignByTimeDataSet(
+ QueryContext context, UDTFPlan udtfPlan, IUDFInputDataSet inputDataSet, boolean keepNull)
+ throws QueryProcessException, IOException {
+ super(context, udtfPlan, inputDataSet);
+ this.keepNull = keepNull;
initTimeHeap();
}
protected void initTimeHeap() throws IOException, QueryProcessException {
timeHeap = new TimeSelector(transformers.length << 1, true);
for (LayerPointReader reader : transformers) {
- // Since a constant operand is not allowed to be a result column, the reader will not be
- // a ConstantLayerPointReader.
- if (reader.next()) {
- timeHeap.add(reader.currentTime());
- }
+ iterateReaderToNextValid(reader);
}
}
@@ -125,58 +132,60 @@ public class UDTFAlignByTimeDataSet extends UDTFDataSet implements DirectAlignBy
}
if (rowOffset == 0) {
- currentBitmapList[i] = (currentBitmapList[i] << 1) | FLAG;
- TSDataType type = reader.getDataType();
- switch (type) {
- case INT32:
- int intValue = reader.currentInt();
- ReadWriteIOUtils.write(
- encoder != null && encoder.needEncode(minTime)
- ? encoder.encodeInt(intValue, minTime)
- : intValue,
- valueBAOSList[i]);
- break;
- case INT64:
- long longValue = reader.currentLong();
- ReadWriteIOUtils.write(
- encoder != null && encoder.needEncode(minTime)
- ? encoder.encodeLong(longValue, minTime)
- : longValue,
- valueBAOSList[i]);
- break;
- case FLOAT:
- float floatValue = reader.currentFloat();
- ReadWriteIOUtils.write(
- encoder != null && encoder.needEncode(minTime)
- ? encoder.encodeFloat(floatValue, minTime)
- : floatValue,
- valueBAOSList[i]);
- break;
- case DOUBLE:
- double doubleValue = reader.currentDouble();
- ReadWriteIOUtils.write(
- encoder != null && encoder.needEncode(minTime)
- ? encoder.encodeDouble(doubleValue, minTime)
- : doubleValue,
- valueBAOSList[i]);
- break;
- case BOOLEAN:
- ReadWriteIOUtils.write(reader.currentBoolean(), valueBAOSList[i]);
- break;
- case TEXT:
- ReadWriteIOUtils.write(reader.currentBinary(), valueBAOSList[i]);
- break;
- default:
- throw new UnSupportedDataTypeException(
- String.format("Data type %s is not supported.", type));
+ if (!reader.isCurrentNull()) {
+ currentBitmapList[i] = (currentBitmapList[i] << 1) | FLAG;
+ TSDataType type = reader.getDataType();
+ switch (type) {
+ case INT32:
+ int intValue = reader.currentInt();
+ ReadWriteIOUtils.write(
+ encoder != null && encoder.needEncode(minTime)
+ ? encoder.encodeInt(intValue, minTime)
+ : intValue,
+ valueBAOSList[i]);
+ break;
+ case INT64:
+ long longValue = reader.currentLong();
+ ReadWriteIOUtils.write(
+ encoder != null && encoder.needEncode(minTime)
+ ? encoder.encodeLong(longValue, minTime)
+ : longValue,
+ valueBAOSList[i]);
+ break;
+ case FLOAT:
+ float floatValue = reader.currentFloat();
+ ReadWriteIOUtils.write(
+ encoder != null && encoder.needEncode(minTime)
+ ? encoder.encodeFloat(floatValue, minTime)
+ : floatValue,
+ valueBAOSList[i]);
+ break;
+ case DOUBLE:
+ double doubleValue = reader.currentDouble();
+ ReadWriteIOUtils.write(
+ encoder != null && encoder.needEncode(minTime)
+ ? encoder.encodeDouble(doubleValue, minTime)
+ : doubleValue,
+ valueBAOSList[i]);
+ break;
+ case BOOLEAN:
+ ReadWriteIOUtils.write(reader.currentBoolean(), valueBAOSList[i]);
+ break;
+ case TEXT:
+ ReadWriteIOUtils.write(reader.currentBinary(), valueBAOSList[i]);
+ break;
+ default:
+ throw new UnSupportedDataTypeException(
+ String.format("Data type %s is not supported.", type));
+ }
+ } else {
+ // there's no data in the current field, so put a null placeholder 0x00
+ currentBitmapList[i] = (currentBitmapList[i] << 1);
}
}
reader.readyForNext();
-
- if (reader.next()) {
- timeHeap.add(reader.currentTime());
- }
+ iterateReaderToNextValid(reader);
}
if (rowOffset == 0) {
@@ -262,35 +271,37 @@ public class UDTFAlignByTimeDataSet extends UDTFDataSet implements DirectAlignBy
rowRecord.addField(null);
continue;
}
- Object value;
- switch (reader.getDataType()) {
- case INT32:
- value = reader.currentInt();
- break;
- case INT64:
- value = reader.currentLong();
- break;
- case FLOAT:
- value = reader.currentFloat();
- break;
- case DOUBLE:
- value = reader.currentDouble();
- break;
- case BOOLEAN:
- value = reader.currentBoolean();
- break;
- case TEXT:
- value = reader.currentBinary();
- break;
- default:
- throw new UnSupportedDataTypeException("Unsupported data type.");
+ if (reader.isCurrentNull()) {
+ rowRecord.addField(null);
+ } else {
+ Object value;
+ switch (reader.getDataType()) {
+ case INT32:
+ value = reader.currentInt();
+ break;
+ case INT64:
+ value = reader.currentLong();
+ break;
+ case FLOAT:
+ value = reader.currentFloat();
+ break;
+ case DOUBLE:
+ value = reader.currentDouble();
+ break;
+ case BOOLEAN:
+ value = reader.currentBoolean();
+ break;
+ case TEXT:
+ value = reader.currentBinary();
+ break;
+ default:
+ throw new UnSupportedDataTypeException("Unsupported data type.");
+ }
+ rowRecord.addField(value, reader.getDataType());
}
- rowRecord.addField(value, reader.getDataType());
reader.readyForNext();
- if (reader.next()) {
- timeHeap.add(reader.currentTime());
- }
+ iterateReaderToNextValid(reader);
}
} catch (QueryProcessException e) {
throw new IOException(e.getMessage());
@@ -300,4 +311,19 @@ public class UDTFAlignByTimeDataSet extends UDTFDataSet implements DirectAlignBy
return rowRecord;
}
+
+ private void iterateReaderToNextValid(LayerPointReader reader)
+ throws QueryProcessException, IOException {
+ // Since a constant operand is not allowed to be a result column, the reader will not be
+ // a ConstantLayerPointReader.
+ // If keepNull is false, we must iterate the reader until a non-null row is returned.
+ while (reader.next()) {
+ if (reader.isCurrentNull() && !keepNull) {
+ reader.readyForNext();
+ continue;
+ }
+ timeHeap.add(reader.currentTime());
+ break;
+ }
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/query/dataset/UDTFDataSet.java b/server/src/main/java/org/apache/iotdb/db/query/dataset/UDTFDataSet.java
index 5a84340..69ed849 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/dataset/UDTFDataSet.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/dataset/UDTFDataSet.java
@@ -102,6 +102,14 @@ public abstract class UDTFDataSet extends QueryDataSet {
initTransformers();
}
+ protected UDTFDataSet(QueryContext queryContext, UDTFPlan udtfPlan, IUDFInputDataSet dataSet)
+ throws QueryProcessException, IOException {
+ queryId = queryContext.getQueryId();
+ this.udtfPlan = udtfPlan;
+ rawQueryInputLayer = new RawQueryInputLayer(queryId, UDF_READER_MEMORY_BUDGET_IN_MB, dataSet);
+ initTransformers();
+ }
+
protected void initTransformers() throws QueryProcessException, IOException {
UDFRegistrationService.getInstance().acquireRegistrationLock();
// This statement must be surrounded by the registration lock.
diff --git a/server/src/main/java/org/apache/iotdb/db/query/dataset/UDTFNonAlignDataSet.java b/server/src/main/java/org/apache/iotdb/db/query/dataset/UDTFNonAlignDataSet.java
index 034c3ad..afd1824 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/dataset/UDTFNonAlignDataSet.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/dataset/UDTFNonAlignDataSet.java
@@ -133,7 +133,7 @@ public class UDTFNonAlignDataSet extends UDTFDataSet implements DirectNonAlignDa
&& (rowLimit <= 0 || alreadyReturnedRowNumArray[transformedDataColumnIndex] < rowLimit)
&& reader.next()) {
- if (offsetArray[transformedDataColumnIndex] == 0) {
+ if (offsetArray[transformedDataColumnIndex] == 0 && !reader.isCurrentNull()) {
long timestamp = reader.currentTime();
ReadWriteIOUtils.write(timestamp, timeBAOS);
diff --git a/server/src/main/java/org/apache/iotdb/db/query/executor/IQueryRouter.java b/server/src/main/java/org/apache/iotdb/db/query/executor/IQueryRouter.java
index aef7399..c0e4461 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/executor/IQueryRouter.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/executor/IQueryRouter.java
@@ -26,6 +26,7 @@ import org.apache.iotdb.db.qp.physical.crud.GroupByTimeFillPlan;
import org.apache.iotdb.db.qp.physical.crud.GroupByTimePlan;
import org.apache.iotdb.db.qp.physical.crud.LastQueryPlan;
import org.apache.iotdb.db.qp.physical.crud.RawDataQueryPlan;
+import org.apache.iotdb.db.qp.physical.crud.UDAFPlan;
import org.apache.iotdb.db.qp.physical.crud.UDTFPlan;
import org.apache.iotdb.db.query.context.QueryContext;
import org.apache.iotdb.tsfile.exception.filter.QueryFilterOptimizationException;
@@ -44,6 +45,11 @@ public interface IQueryRouter {
throws QueryFilterOptimizationException, StorageEngineException, IOException,
QueryProcessException;
+ /** Execute UDAF query. */
+ QueryDataSet udafQuery(UDAFPlan udafPlan, QueryContext context)
+ throws QueryFilterOptimizationException, StorageEngineException, IOException,
+ QueryProcessException;
+
/** Execute groupBy query. */
QueryDataSet groupBy(GroupByTimePlan groupByTimePlan, QueryContext context)
throws QueryFilterOptimizationException, StorageEngineException, QueryProcessException,
diff --git a/server/src/main/java/org/apache/iotdb/db/query/executor/QueryRouter.java b/server/src/main/java/org/apache/iotdb/db/query/executor/QueryRouter.java
index 017780c..033186d 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/executor/QueryRouter.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/executor/QueryRouter.java
@@ -28,6 +28,7 @@ import org.apache.iotdb.db.qp.physical.crud.GroupByTimeFillPlan;
import org.apache.iotdb.db.qp.physical.crud.GroupByTimePlan;
import org.apache.iotdb.db.qp.physical.crud.LastQueryPlan;
import org.apache.iotdb.db.qp.physical.crud.RawDataQueryPlan;
+import org.apache.iotdb.db.qp.physical.crud.UDAFPlan;
import org.apache.iotdb.db.qp.physical.crud.UDTFPlan;
import org.apache.iotdb.db.query.context.QueryContext;
import org.apache.iotdb.db.query.control.SessionManager;
@@ -39,7 +40,9 @@ import org.apache.iotdb.db.query.dataset.groupby.GroupByLevelDataSet;
import org.apache.iotdb.db.query.dataset.groupby.GroupByWithValueFilterDataSet;
import org.apache.iotdb.db.query.dataset.groupby.GroupByWithoutValueFilterDataSet;
import org.apache.iotdb.db.utils.TimeValuePairUtils;
+import org.apache.iotdb.db.utils.TypeInferenceUtils;
import org.apache.iotdb.tsfile.exception.filter.QueryFilterOptimizationException;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.read.expression.ExpressionType;
import org.apache.iotdb.tsfile.read.expression.IExpression;
import org.apache.iotdb.tsfile.read.expression.impl.BinaryExpression;
@@ -156,6 +159,36 @@ public class QueryRouter implements IQueryRouter {
return dataSet;
}
+ @Override
+ public QueryDataSet udafQuery(UDAFPlan udafPlan, QueryContext context)
+ throws QueryFilterOptimizationException, StorageEngineException, IOException,
+ QueryProcessException {
+ if (logger.isDebugEnabled()) {
+ logger.debug("paths:" + udafPlan.getPaths());
+ }
+ AggregationPlan innerPlan = udafPlan.getInnerAggregationPlan();
+ // Infer aggregation data types for UDF input
+ List<TSDataType> aggregationResultTypes = new ArrayList<>();
+ for (int i = 0; i < innerPlan.getDeduplicatedPaths().size(); i++) {
+ aggregationResultTypes.add(
+ TypeInferenceUtils.getAggrDataType(
+ innerPlan.getDeduplicatedAggregations().get(i),
+ innerPlan.getDeduplicatedDataTypes().get(i)));
+ }
+ QueryDataSet innerQueryDataSet = null;
+ boolean keepNull = false;
+ if (innerPlan instanceof GroupByTimePlan) {
+ innerQueryDataSet = groupBy((GroupByTimePlan) innerPlan, context);
+ keepNull = true;
+ } else {
+ innerQueryDataSet = aggregate(innerPlan, context);
+ }
+
+ UDFQueryExecutor udfQueryExecutor = new UDFQueryExecutor(udafPlan);
+ return udfQueryExecutor.executeFromAlignedDataSet(
+ context, innerQueryDataSet, aggregationResultTypes, keepNull);
+ }
+
protected AggregationExecutor getAggregationExecutor(
QueryContext context, AggregationPlan aggregationPlan) {
return new AggregationExecutor(context, aggregationPlan);
@@ -316,7 +349,7 @@ public class QueryRouter implements IQueryRouter {
boolean withValueFilter =
optimizedExpression != null && optimizedExpression.getType() != ExpressionType.GLOBAL_TIME;
- UDTFQueryExecutor udtfQueryExecutor = new UDTFQueryExecutor(udtfPlan);
+ UDFQueryExecutor udtfQueryExecutor = new UDFQueryExecutor(udtfPlan);
if (udtfPlan.isAlignByTime()) {
return withValueFilter
diff --git a/server/src/main/java/org/apache/iotdb/db/query/executor/UDTFQueryExecutor.java b/server/src/main/java/org/apache/iotdb/db/query/executor/UDFQueryExecutor.java
similarity index 86%
rename from server/src/main/java/org/apache/iotdb/db/query/executor/UDTFQueryExecutor.java
rename to server/src/main/java/org/apache/iotdb/db/query/executor/UDFQueryExecutor.java
index 934ec2d..0349616 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/executor/UDTFQueryExecutor.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/executor/UDFQueryExecutor.java
@@ -23,10 +23,12 @@ import org.apache.iotdb.db.exception.StorageEngineException;
import org.apache.iotdb.db.exception.query.QueryProcessException;
import org.apache.iotdb.db.qp.physical.crud.UDTFPlan;
import org.apache.iotdb.db.query.context.QueryContext;
+import org.apache.iotdb.db.query.dataset.UDFInputDataSet;
import org.apache.iotdb.db.query.dataset.UDTFAlignByTimeDataSet;
import org.apache.iotdb.db.query.dataset.UDTFNonAlignDataSet;
import org.apache.iotdb.db.query.reader.series.IReaderByTimestamp;
import org.apache.iotdb.db.query.reader.series.ManagedSeriesReader;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.read.query.dataset.QueryDataSet;
import org.apache.iotdb.tsfile.read.query.timegenerator.TimeGenerator;
@@ -36,11 +38,11 @@ import java.util.List;
import static org.apache.iotdb.tsfile.read.query.executor.ExecutorWithTimeGenerator.markFilterdPaths;
-public class UDTFQueryExecutor extends RawDataQueryExecutor {
+public class UDFQueryExecutor extends RawDataQueryExecutor {
protected final UDTFPlan udtfPlan;
- public UDTFQueryExecutor(UDTFPlan udtfPlan) {
+ public UDFQueryExecutor(UDTFPlan udtfPlan) {
super(udtfPlan);
this.udtfPlan = udtfPlan;
}
@@ -84,4 +86,14 @@ public class UDTFQueryExecutor extends RawDataQueryExecutor {
return new UDTFNonAlignDataSet(
context, udtfPlan, timestampGenerator, readersOfSelectedSeries, cached);
}
+
+ public final QueryDataSet executeFromAlignedDataSet(
+ QueryContext context,
+ QueryDataSet sourceDataSet,
+ List<TSDataType> fieldTypes,
+ boolean keepNull)
+ throws QueryProcessException, IOException {
+ return new UDTFAlignByTimeDataSet(
+ context, udtfPlan, new UDFInputDataSet(sourceDataSet, fieldTypes), keepNull);
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/query/expression/Expression.java b/server/src/main/java/org/apache/iotdb/db/query/expression/Expression.java
index 5558b50..0bf537d 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/expression/Expression.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/expression/Expression.java
@@ -33,6 +33,9 @@ import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import java.io.IOException;
import java.time.ZoneId;
+import java.util.Deque;
+import java.util.Iterator;
+import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Set;
@@ -47,6 +50,10 @@ public abstract class Expression {
return false;
}
+ public boolean isUDAFExpression() {
+ return false;
+ }
+
public boolean isTimeSeriesGeneratingFunctionExpression() {
return false;
}
@@ -76,6 +83,11 @@ public abstract class Expression {
/** Sub-classes should override this method indicating if the expression is a constant operand */
protected abstract boolean isConstantOperandInternal();
+ /**
+ * returns the DIRECT children expressions if it has any, otherwise an EMPTY list will be returned
+ */
+ public abstract List<Expression> getExpressions();
+
/** If this expression and all of its sub-expressions are {@link ConstantOperand}. */
public final boolean isConstantOperand() {
if (isConstantOperandCache == null) {
@@ -128,4 +140,38 @@ public abstract class Expression {
public final String toString() {
return getExpressionString();
}
+
+ /** returns an iterator to traverse all the successor expressions in a level-order */
+ public final Iterator<Expression> iterator() {
+ return new ExpressionIterator(this);
+ }
+
+ /** the iterator of an Expression tree with level-order traversal */
+ private static class ExpressionIterator implements Iterator<Expression> {
+
+ private final Deque<Expression> queue = new LinkedList<>();
+
+ public ExpressionIterator(Expression expression) {
+ queue.add(expression);
+ }
+
+ @Override
+ public boolean hasNext() {
+ return !queue.isEmpty();
+ }
+
+ @Override
+ public Expression next() {
+ if (!hasNext()) {
+ return null;
+ }
+ Expression current = queue.pop();
+ if (current != null) {
+ for (Expression subExp : current.getExpressions()) {
+ queue.push(subExp);
+ }
+ }
+ return current;
+ }
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/query/expression/ResultColumn.java b/server/src/main/java/org/apache/iotdb/db/query/expression/ResultColumn.java
index 8e526ab..8be34eb 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/expression/ResultColumn.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/expression/ResultColumn.java
@@ -152,4 +152,21 @@ public class ResultColumn {
public TSDataType getDataType() {
return dataType;
}
+
+ @Override
+ public final int hashCode() {
+ return alias == null ? getResultColumnName().hashCode() : alias.hashCode();
+ }
+
+ @Override
+ public final boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+
+ if (!(o instanceof ResultColumn)) {
+ return false;
+ }
+ return getResultColumnName().equals(((ResultColumn) o).getResultColumnName());
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/query/expression/binary/BinaryExpression.java b/server/src/main/java/org/apache/iotdb/db/query/expression/binary/BinaryExpression.java
index df55256..f5e1d23 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/expression/binary/BinaryExpression.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/expression/binary/BinaryExpression.java
@@ -39,6 +39,7 @@ import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import java.io.IOException;
import java.time.ZoneId;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.Set;
@@ -53,6 +54,14 @@ public abstract class BinaryExpression extends Expression {
this.rightExpression = rightExpression;
}
+ public Expression getLeftExpression() {
+ return leftExpression;
+ }
+
+ public Expression getRightExpression() {
+ return rightExpression;
+ }
+
@Override
public boolean isConstantOperandInternal() {
return leftExpression.isConstantOperand() && rightExpression.isConstantOperand();
@@ -60,7 +69,20 @@ public abstract class BinaryExpression extends Expression {
@Override
public boolean isTimeSeriesGeneratingFunctionExpression() {
- return true;
+ return !isUDAFExpression();
+ }
+
+ @Override
+ public boolean isUDAFExpression() {
+ return leftExpression.isAggregationFunctionExpression()
+ || rightExpression.isAggregationFunctionExpression()
+ || leftExpression.isUDAFExpression()
+ || rightExpression.isUDAFExpression();
+ }
+
+ @Override
+ public List<Expression> getExpressions() {
+ return Arrays.asList(leftExpression, rightExpression);
}
@Override
diff --git a/server/src/main/java/org/apache/iotdb/db/query/expression/unary/ConstantOperand.java b/server/src/main/java/org/apache/iotdb/db/query/expression/unary/ConstantOperand.java
index cb8f61d..5f838ec 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/expression/unary/ConstantOperand.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/expression/unary/ConstantOperand.java
@@ -34,6 +34,7 @@ import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.commons.lang3.Validate;
import java.time.ZoneId;
+import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Set;
@@ -75,6 +76,11 @@ public class ConstantOperand extends Expression {
}
@Override
+ public List<Expression> getExpressions() {
+ return Collections.emptyList();
+ }
+
+ @Override
public void constructUdfExecutors(
Map<String, UDTFExecutor> expressionName2Executor, ZoneId zoneId) {
// Do nothing
diff --git a/server/src/main/java/org/apache/iotdb/db/query/expression/unary/FunctionExpression.java b/server/src/main/java/org/apache/iotdb/db/query/expression/unary/FunctionExpression.java
index dbd8211..885f2bd 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/expression/unary/FunctionExpression.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/expression/unary/FunctionExpression.java
@@ -37,6 +37,7 @@ import org.apache.iotdb.db.query.udf.core.layer.RawQueryInputLayer;
import org.apache.iotdb.db.query.udf.core.layer.SingleInputColumnMultiReferenceIntermediateLayer;
import org.apache.iotdb.db.query.udf.core.layer.SingleInputColumnSingleReferenceIntermediateLayer;
import org.apache.iotdb.db.query.udf.core.transformer.Transformer;
+import org.apache.iotdb.db.query.udf.core.transformer.TransparentTransformer;
import org.apache.iotdb.db.query.udf.core.transformer.UDFQueryRowTransformer;
import org.apache.iotdb.db.query.udf.core.transformer.UDFQueryRowWindowTransformer;
import org.apache.iotdb.db.query.udf.core.transformer.UDFQueryTransformer;
@@ -61,6 +62,8 @@ public class FunctionExpression extends Expression {
*/
private final boolean isAggregationFunctionExpression;
+ private boolean isUDAFExpression;
+
private final String functionName;
private final Map<String, String> functionAttributes;
@@ -71,8 +74,6 @@ public class FunctionExpression extends Expression {
*/
private List<Expression> expressions;
- private List<TSDataType> inputExpressionTypes;
-
private List<PartialPath> paths;
private String parametersString;
@@ -94,6 +95,9 @@ public class FunctionExpression extends Expression {
isAggregationFunctionExpression =
SQLConstant.getNativeFunctionNames().contains(functionName.toLowerCase());
isConstantOperandCache = expressions.stream().anyMatch(Expression::isConstantOperand);
+ isUDAFExpression =
+ expressions.stream()
+ .anyMatch(v -> v.isUDAFExpression() || v.isAggregationFunctionExpression());
}
@Override
@@ -107,6 +111,11 @@ public class FunctionExpression extends Expression {
}
@Override
+ public boolean isUDAFExpression() {
+ return isUDAFExpression;
+ }
+
+ @Override
public boolean isTimeSeriesGeneratingFunctionExpression() {
return !isAggregationFunctionExpression;
}
@@ -123,6 +132,10 @@ public class FunctionExpression extends Expression {
public void addExpression(Expression expression) {
isConstantOperandCache = isConstantOperandCache && expression.isConstantOperand();
+ isUDAFExpression =
+ isUDAFExpression
+ || expression.isUDAFExpression()
+ || expression.isAggregationFunctionExpression();
expressions.add(expression);
}
@@ -210,20 +223,30 @@ public class FunctionExpression extends Expression {
throws QueryProcessException, IOException {
if (!expressionIntermediateLayerMap.containsKey(this)) {
float memoryBudgetInMB = memoryAssigner.assign();
-
- IntermediateLayer udfInputIntermediateLayer =
- constructUdfInputIntermediateLayer(
- queryId,
- udtfPlan,
- rawTimeSeriesInputLayer,
- expressionIntermediateLayerMap,
- expressionDataTypeMap,
- memoryAssigner);
- Transformer transformer =
- constructUdfTransformer(
- queryId, udtfPlan, expressionDataTypeMap, memoryAssigner, udfInputIntermediateLayer);
+ Transformer transformer;
+ if (isAggregationFunctionExpression) {
+ transformer =
+ new TransparentTransformer(
+ rawTimeSeriesInputLayer.constructPointReader(
+ udtfPlan.getReaderIndexByExpressionName(toString())));
+ } else {
+ IntermediateLayer udfInputIntermediateLayer =
+ constructUdfInputIntermediateLayer(
+ queryId,
+ udtfPlan,
+ rawTimeSeriesInputLayer,
+ expressionIntermediateLayerMap,
+ expressionDataTypeMap,
+ memoryAssigner);
+ transformer =
+ constructUdfTransformer(
+ queryId,
+ udtfPlan,
+ expressionDataTypeMap,
+ memoryAssigner,
+ udfInputIntermediateLayer);
+ }
expressionDataTypeMap.put(this, transformer.getDataType());
-
expressionIntermediateLayerMap.put(
this,
memoryAssigner.getReference(this) == 1
diff --git a/server/src/main/java/org/apache/iotdb/db/query/expression/unary/NegationExpression.java b/server/src/main/java/org/apache/iotdb/db/query/expression/unary/NegationExpression.java
index 2a514ba..ced0532 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/expression/unary/NegationExpression.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/expression/unary/NegationExpression.java
@@ -38,6 +38,7 @@ import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import java.io.IOException;
import java.time.ZoneId;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Set;
@@ -60,11 +61,21 @@ public class NegationExpression extends Expression {
}
@Override
+ public List<Expression> getExpressions() {
+ return Collections.singletonList(expression);
+ }
+
+ @Override
public boolean isTimeSeriesGeneratingFunctionExpression() {
return true;
}
@Override
+ public boolean isUDAFExpression() {
+ return expression.isUDAFExpression() || expression.isAggregationFunctionExpression();
+ }
+
+ @Override
public void concat(List<PartialPath> prefixPaths, List<Expression> resultExpressions) {
List<Expression> resultExpressionsForRecursion = new ArrayList<>();
expression.concat(prefixPaths, resultExpressionsForRecursion);
diff --git a/server/src/main/java/org/apache/iotdb/db/query/expression/unary/TimeSeriesOperand.java b/server/src/main/java/org/apache/iotdb/db/query/expression/unary/TimeSeriesOperand.java
index 7f4bb48..a538e09 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/expression/unary/TimeSeriesOperand.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/expression/unary/TimeSeriesOperand.java
@@ -35,6 +35,7 @@ import org.apache.iotdb.db.query.udf.core.reader.LayerPointReader;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import java.time.ZoneId;
+import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Set;
@@ -61,6 +62,11 @@ public class TimeSeriesOperand extends Expression {
}
@Override
+ public List<Expression> getExpressions() {
+ return Collections.emptyList();
+ }
+
+ @Override
public void concat(List<PartialPath> prefixPaths, List<Expression> resultExpressions) {
for (PartialPath prefixPath : prefixPaths) {
resultExpressions.add(new TimeSeriesOperand(prefixPath.concatPath(path)));
diff --git a/server/src/main/java/org/apache/iotdb/db/query/udf/core/executor/UDTFExecutor.java b/server/src/main/java/org/apache/iotdb/db/query/udf/core/executor/UDTFExecutor.java
index 37269d3..14517e6 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/udf/core/executor/UDTFExecutor.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/udf/core/executor/UDTFExecutor.java
@@ -80,9 +80,14 @@ public class UDTFExecutor {
configurations.getOutputDataType(), queryId, collectorMemoryBudgetInMB, 1);
}
- public void execute(Row row) throws QueryProcessException {
+ public void execute(Row row, boolean isCurrentRowNull) throws QueryProcessException {
try {
- udtf.transform(row, collector);
+ if (isCurrentRowNull) {
+ // A null row will never trigger any UDF computing
+ collector.putNull(row.getTime());
+ } else {
+ udtf.transform(row, collector);
+ }
} catch (Exception e) {
onError("transform(Row, PointCollector)", e);
}
@@ -105,7 +110,9 @@ public class UDTFExecutor {
}
public void beforeDestroy() {
- udtf.beforeDestroy();
+ if (udtf != null) {
+ udtf.beforeDestroy();
+ }
}
private void onError(String methodName, Exception e) throws QueryProcessException {
diff --git a/server/src/main/java/org/apache/iotdb/db/query/dataset/UDFInputDataSet.java b/server/src/main/java/org/apache/iotdb/db/query/udf/core/layer/InputRowUtils.java
similarity index 55%
copy from server/src/main/java/org/apache/iotdb/db/query/dataset/UDFInputDataSet.java
copy to server/src/main/java/org/apache/iotdb/db/query/udf/core/layer/InputRowUtils.java
index d404288..afe846b 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/dataset/UDFInputDataSet.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/udf/core/layer/InputRowUtils.java
@@ -16,19 +16,28 @@
* specific language governing permissions and limitations
* under the License.
*/
+package org.apache.iotdb.db.query.udf.core.layer;
-package org.apache.iotdb.db.query.dataset;
+import org.apache.iotdb.db.query.dataset.IUDFInputDataSet;
-import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+public class InputRowUtils {
-import java.io.IOException;
-import java.util.List;
-
-public interface UDFInputDataSet {
-
- List<TSDataType> getDataTypes();
-
- boolean hasNextRowInObjects() throws IOException;
-
- Object[] nextRowInObjects() throws IOException;
+ /**
+ * this method checks whether the row returned by IUDFInputDataSet.nextRowInObjects() has all null
+ * fields except the timestamp
+ *
+ * @param row the returned row by calling {@link IUDFInputDataSet#nextRowInObjects()}
+ * @return true if all row fields are null.
+ */
+ public static boolean isAllNull(Object[] row) {
+ if (row == null) {
+ return true;
+ }
+ for (int i = 0; i < row.length - 1; i++) {
+ if (row[i] != null) {
+ return false;
+ }
+ }
+ return true;
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/query/udf/core/layer/LayerCacheUtils.java b/server/src/main/java/org/apache/iotdb/db/query/udf/core/layer/LayerCacheUtils.java
index d098699..ab63797 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/udf/core/layer/LayerCacheUtils.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/udf/core/layer/LayerCacheUtils.java
@@ -20,7 +20,7 @@
package org.apache.iotdb.db.query.udf.core.layer;
import org.apache.iotdb.db.exception.query.QueryProcessException;
-import org.apache.iotdb.db.query.dataset.UDFInputDataSet;
+import org.apache.iotdb.db.query.dataset.IUDFInputDataSet;
import org.apache.iotdb.db.query.udf.core.reader.LayerPointReader;
import org.apache.iotdb.db.query.udf.datastructure.row.ElasticSerializableRowRecordList;
import org.apache.iotdb.db.query.udf.datastructure.tv.ElasticSerializableTVList;
@@ -52,28 +52,31 @@ public class LayerCacheUtils {
if (!source.next()) {
return false;
}
-
- switch (dataType) {
- case INT32:
- target.putInt(source.currentTime(), source.currentInt());
- break;
- case INT64:
- target.putLong(source.currentTime(), source.currentLong());
- break;
- case FLOAT:
- target.putFloat(source.currentTime(), source.currentFloat());
- break;
- case DOUBLE:
- target.putDouble(source.currentTime(), source.currentDouble());
- break;
- case BOOLEAN:
- target.putBoolean(source.currentTime(), source.currentBoolean());
- break;
- case TEXT:
- target.putBinary(source.currentTime(), source.currentBinary());
- break;
- default:
- throw new UnsupportedOperationException(dataType.name());
+ if (source.isCurrentNull()) {
+ target.putNull(source.currentTime());
+ } else {
+ switch (dataType) {
+ case INT32:
+ target.putInt(source.currentTime(), source.currentInt());
+ break;
+ case INT64:
+ target.putLong(source.currentTime(), source.currentLong());
+ break;
+ case FLOAT:
+ target.putFloat(source.currentTime(), source.currentFloat());
+ break;
+ case DOUBLE:
+ target.putDouble(source.currentTime(), source.currentDouble());
+ break;
+ case BOOLEAN:
+ target.putBoolean(source.currentTime(), source.currentBoolean());
+ break;
+ case TEXT:
+ target.putBinary(source.currentTime(), source.currentBinary());
+ break;
+ default:
+ throw new UnsupportedOperationException(dataType.name());
+ }
}
source.readyForNext();
@@ -83,7 +86,7 @@ public class LayerCacheUtils {
/** @return number of actually collected, which may be less than or equals to rowsNumber */
public static int cacheRows(
- UDFInputDataSet source, ElasticSerializableRowRecordList target, int rowsNumber)
+ IUDFInputDataSet source, ElasticSerializableRowRecordList target, int rowsNumber)
throws QueryProcessException, IOException {
int count = 0;
while (count < rowsNumber && cacheRow(source, target)) {
@@ -92,7 +95,7 @@ public class LayerCacheUtils {
return count;
}
- public static boolean cacheRow(UDFInputDataSet source, ElasticSerializableRowRecordList target)
+ public static boolean cacheRow(IUDFInputDataSet source, ElasticSerializableRowRecordList target)
throws IOException, QueryProcessException {
if (source.hasNextRowInObjects()) {
target.put(source.nextRowInObjects());
diff --git a/server/src/main/java/org/apache/iotdb/db/query/udf/core/layer/MultiInputColumnIntermediateLayer.java b/server/src/main/java/org/apache/iotdb/db/query/udf/core/layer/MultiInputColumnIntermediateLayer.java
index f36511f..f920424 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/udf/core/layer/MultiInputColumnIntermediateLayer.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/udf/core/layer/MultiInputColumnIntermediateLayer.java
@@ -20,7 +20,7 @@
package org.apache.iotdb.db.query.udf.core.layer;
import org.apache.iotdb.db.exception.query.QueryProcessException;
-import org.apache.iotdb.db.query.dataset.UDFInputDataSet;
+import org.apache.iotdb.db.query.dataset.IUDFInputDataSet;
import org.apache.iotdb.db.query.expression.Expression;
import org.apache.iotdb.db.query.udf.api.access.Row;
import org.apache.iotdb.db.query.udf.api.access.RowWindow;
@@ -41,7 +41,7 @@ import java.util.Arrays;
import java.util.List;
public class MultiInputColumnIntermediateLayer extends IntermediateLayer
- implements UDFInputDataSet {
+ implements IUDFInputDataSet {
private final LayerPointReader[] layerPointReaders;
private final TSDataType[] dataTypes;
@@ -99,28 +99,31 @@ public class MultiInputColumnIntermediateLayer extends IntermediateLayer
continue;
}
- switch (reader.getDataType()) {
- case INT32:
- row[i] = reader.currentInt();
- break;
- case INT64:
- row[i] = reader.currentLong();
- break;
- case FLOAT:
- row[i] = reader.currentFloat();
- break;
- case DOUBLE:
- row[i] = reader.currentDouble();
- break;
- case BOOLEAN:
- row[i] = reader.currentBoolean();
- break;
- case TEXT:
- row[i] = reader.currentBinary();
- break;
- default:
- throw new UnSupportedDataTypeException("Unsupported data type.");
+ if (!reader.isCurrentNull()) {
+ switch (reader.getDataType()) {
+ case INT32:
+ row[i] = reader.currentInt();
+ break;
+ case INT64:
+ row[i] = reader.currentLong();
+ break;
+ case FLOAT:
+ row[i] = reader.currentFloat();
+ break;
+ case DOUBLE:
+ row[i] = reader.currentDouble();
+ break;
+ case BOOLEAN:
+ row[i] = reader.currentBoolean();
+ break;
+ case TEXT:
+ row[i] = reader.currentBinary();
+ break;
+ default:
+ throw new UnSupportedDataTypeException("Unsupported data type.");
+ }
}
+
reader.readyForNext();
if (!(reader.isConstantPointReader()) && reader.next()) {
@@ -148,6 +151,7 @@ public class MultiInputColumnIntermediateLayer extends IntermediateLayer
new ElasticSerializableRowRecordListBackedMultiColumnRow(dataTypes);
private boolean hasCached = false;
+ private boolean currentNull = false;
@Override
public boolean next() throws IOException {
@@ -158,8 +162,9 @@ public class MultiInputColumnIntermediateLayer extends IntermediateLayer
if (!hasNextRowInObjects()) {
return false;
}
-
- row.setRowRecord(nextRowInObjects());
+ Object[] rowRecords = nextRowInObjects();
+ currentNull = InputRowUtils.isAllNull(rowRecords);
+ row.setRowRecord(rowRecords);
hasCached = true;
return true;
}
@@ -167,6 +172,7 @@ public class MultiInputColumnIntermediateLayer extends IntermediateLayer
@Override
public void readyForNext() {
hasCached = false;
+ currentNull = false;
}
@Override
@@ -183,6 +189,11 @@ public class MultiInputColumnIntermediateLayer extends IntermediateLayer
public Row currentRow() {
return row;
}
+
+ @Override
+ public boolean isCurrentNull() {
+ return currentNull;
+ }
};
}
@@ -191,7 +202,7 @@ public class MultiInputColumnIntermediateLayer extends IntermediateLayer
SlidingSizeWindowAccessStrategy strategy, float memoryBudgetInMB)
throws QueryProcessException {
- final UDFInputDataSet udfInputDataSet = this;
+ final IUDFInputDataSet udfInputDataSet = this;
return new LayerRowWindowReader() {
@@ -260,7 +271,7 @@ public class MultiInputColumnIntermediateLayer extends IntermediateLayer
final long slidingStep = strategy.getSlidingStep();
final long displayWindowEnd = strategy.getDisplayWindowEnd();
- final UDFInputDataSet udfInputDataSet = this;
+ final IUDFInputDataSet udfInputDataSet = this;
final ElasticSerializableRowRecordList rowRecordList =
new ElasticSerializableRowRecordList(
dataTypes, queryId, memoryBudgetInMB, CACHE_BLOCK_SIZE);
diff --git a/server/src/main/java/org/apache/iotdb/db/query/udf/core/layer/RawQueryInputLayer.java b/server/src/main/java/org/apache/iotdb/db/query/udf/core/layer/RawQueryInputLayer.java
index e85b783..b60e37c 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/udf/core/layer/RawQueryInputLayer.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/udf/core/layer/RawQueryInputLayer.java
@@ -21,8 +21,8 @@ package org.apache.iotdb.db.query.udf.core.layer;
import org.apache.iotdb.db.exception.query.QueryProcessException;
import org.apache.iotdb.db.metadata.path.PartialPath;
+import org.apache.iotdb.db.query.dataset.IUDFInputDataSet;
import org.apache.iotdb.db.query.dataset.RawQueryDataSetWithValueFilter;
-import org.apache.iotdb.db.query.dataset.UDFInputDataSet;
import org.apache.iotdb.db.query.dataset.UDFRawQueryInputDataSetWithoutValueFilter;
import org.apache.iotdb.db.query.reader.series.IReaderByTimestamp;
import org.apache.iotdb.db.query.reader.series.ManagedSeriesReader;
@@ -38,7 +38,7 @@ import java.util.List;
public class RawQueryInputLayer {
- private UDFInputDataSet queryDataSet;
+ private IUDFInputDataSet queryDataSet;
private TSDataType[] dataTypes;
private int timestampIndex;
@@ -75,7 +75,12 @@ public class RawQueryInputLayer {
new RawQueryDataSetWithValueFilter(paths, dataTypes, timeGenerator, readers, cached, true));
}
- private void construct(long queryId, float memoryBudgetInMB, UDFInputDataSet queryDataSet)
+ public RawQueryInputLayer(long queryId, float memoryBudgetInMB, IUDFInputDataSet queryDataSet)
+ throws QueryProcessException {
+ construct(queryId, memoryBudgetInMB, queryDataSet);
+ }
+
+ private void construct(long queryId, float memoryBudgetInMB, IUDFInputDataSet queryDataSet)
throws QueryProcessException {
this.queryDataSet = queryDataSet;
dataTypes = queryDataSet.getDataTypes().toArray(new TSDataType[0]);
@@ -127,7 +132,11 @@ public class RawQueryInputLayer {
for (int i = currentRowIndex + 1; i < rowRecordList.size(); ++i) {
Object[] rowRecordCandidate = rowRecordList.getRowRecord(i);
- if (rowRecordCandidate[columnIndex] != null) {
+ // It all fields except the timestamp in the current row are null, we should treat this row
+ // be valid. Because in a GROUP BY time query, we must return every time window record even
+ // if there's no data. Under the situation, if hasCachedRowRecord is false, this row will be
+ // skipped and the result is not as our expected.
+ if (rowRecordCandidate[columnIndex] != null || rowRecordList.fieldsAllNull(i)) {
hasCachedRowRecord = true;
cachedRowRecord = rowRecordCandidate;
currentRowIndex = i;
@@ -139,7 +148,8 @@ public class RawQueryInputLayer {
while (queryDataSet.hasNextRowInObjects()) {
Object[] rowRecordCandidate = queryDataSet.nextRowInObjects();
rowRecordList.put(rowRecordCandidate);
- if (rowRecordCandidate[columnIndex] != null) {
+ if (rowRecordCandidate[columnIndex] != null
+ || rowRecordList.fieldsAllNull(rowRecordList.size() - 1)) {
hasCachedRowRecord = true;
cachedRowRecord = rowRecordCandidate;
currentRowIndex = rowRecordList.size() - 1;
@@ -195,6 +205,11 @@ public class RawQueryInputLayer {
}
@Override
+ public boolean isCurrentNull() {
+ return cachedRowRecord[columnIndex] == null;
+ }
+
+ @Override
public Binary currentBinary() {
return (Binary) cachedRowRecord[columnIndex];
}
diff --git a/server/src/main/java/org/apache/iotdb/db/query/udf/core/layer/SingleInputColumnMultiReferenceIntermediateLayer.java b/server/src/main/java/org/apache/iotdb/db/query/udf/core/layer/SingleInputColumnMultiReferenceIntermediateLayer.java
index 03636ed..7c56de5 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/udf/core/layer/SingleInputColumnMultiReferenceIntermediateLayer.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/udf/core/layer/SingleInputColumnMultiReferenceIntermediateLayer.java
@@ -134,6 +134,11 @@ public class SingleInputColumnMultiReferenceIntermediateLayer extends Intermedia
public Binary currentBinary() throws IOException {
return tvList.getBinary(currentPointIndex);
}
+
+ @Override
+ public boolean isCurrentNull() throws IOException {
+ return tvList.isNull(currentPointIndex);
+ }
};
}
@@ -183,6 +188,11 @@ public class SingleInputColumnMultiReferenceIntermediateLayer extends Intermedia
public Row currentRow() {
return row;
}
+
+ @Override
+ public boolean isCurrentNull() throws IOException {
+ return tvList.isNull(currentRowIndex);
+ }
};
}
diff --git a/server/src/main/java/org/apache/iotdb/db/query/udf/core/layer/SingleInputColumnSingleReferenceIntermediateLayer.java b/server/src/main/java/org/apache/iotdb/db/query/udf/core/layer/SingleInputColumnSingleReferenceIntermediateLayer.java
index be5d41c..c4ede3e 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/udf/core/layer/SingleInputColumnSingleReferenceIntermediateLayer.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/udf/core/layer/SingleInputColumnSingleReferenceIntermediateLayer.java
@@ -63,11 +63,15 @@ public class SingleInputColumnSingleReferenceIntermediateLayer extends Intermedi
private final Row row = new LayerPointReaderBackedSingleColumnRow(parentLayerPointReader);
private boolean hasCached = false;
+ private boolean isCurrentNull = false;
@Override
public boolean next() throws IOException, QueryProcessException {
if (!hasCached) {
hasCached = parentLayerPointReader.next();
+ if (hasCached) {
+ isCurrentNull = parentLayerPointReader.isCurrentNull();
+ }
}
return hasCached;
}
@@ -75,7 +79,7 @@ public class SingleInputColumnSingleReferenceIntermediateLayer extends Intermedi
@Override
public void readyForNext() {
hasCached = false;
-
+ isCurrentNull = false;
parentLayerPointReader.readyForNext();
}
@@ -93,6 +97,11 @@ public class SingleInputColumnSingleReferenceIntermediateLayer extends Intermedi
public Row currentRow() {
return row;
}
+
+ @Override
+ public boolean isCurrentNull() {
+ return isCurrentNull;
+ }
};
}
diff --git a/server/src/main/java/org/apache/iotdb/db/query/udf/core/reader/ConstantLayerPointReader.java b/server/src/main/java/org/apache/iotdb/db/query/udf/core/reader/ConstantLayerPointReader.java
index 9e54989..058d606 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/udf/core/reader/ConstantLayerPointReader.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/udf/core/reader/ConstantLayerPointReader.java
@@ -129,4 +129,9 @@ public class ConstantLayerPointReader implements LayerPointReader {
public Binary currentBinary() throws IOException {
return cachedBinary;
}
+
+ @Override
+ public boolean isCurrentNull() {
+ return false;
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/query/udf/core/reader/LayerPointReader.java b/server/src/main/java/org/apache/iotdb/db/query/udf/core/reader/LayerPointReader.java
index 9b7a27b..3ee7c5a 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/udf/core/reader/LayerPointReader.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/udf/core/reader/LayerPointReader.java
@@ -47,5 +47,7 @@ public interface LayerPointReader {
boolean currentBoolean() throws IOException;
+ boolean isCurrentNull() throws IOException;
+
Binary currentBinary() throws IOException;
}
diff --git a/server/src/main/java/org/apache/iotdb/db/query/udf/core/reader/LayerRowReader.java b/server/src/main/java/org/apache/iotdb/db/query/udf/core/reader/LayerRowReader.java
index 0bc6521..e2a2f23 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/udf/core/reader/LayerRowReader.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/udf/core/reader/LayerRowReader.java
@@ -36,4 +36,7 @@ public interface LayerRowReader {
long currentTime() throws IOException;
Row currentRow();
+
+ /** whether current row fields are all null */
+ boolean isCurrentNull() throws IOException;
}
diff --git a/server/src/main/java/org/apache/iotdb/db/query/udf/core/transformer/ArithmeticBinaryTransformer.java b/server/src/main/java/org/apache/iotdb/db/query/udf/core/transformer/ArithmeticBinaryTransformer.java
index 76c8595..b93e48a 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/udf/core/transformer/ArithmeticBinaryTransformer.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/udf/core/transformer/ArithmeticBinaryTransformer.java
@@ -49,10 +49,14 @@ public abstract class ArithmeticBinaryTransformer extends Transformer {
if (!cacheTime()) {
return false;
}
- cachedDouble =
- evaluate(
- castCurrentValueToDoubleOperand(leftPointReader),
- castCurrentValueToDoubleOperand(rightPointReader));
+ if (leftPointReader.isCurrentNull() || rightPointReader.isCurrentNull()) {
+ currentNull = true;
+ } else {
+ cachedDouble =
+ evaluate(
+ castCurrentValueToDoubleOperand(leftPointReader),
+ castCurrentValueToDoubleOperand(rightPointReader));
+ }
leftPointReader.readyForNext();
rightPointReader.readyForNext();
return true;
diff --git a/server/src/main/java/org/apache/iotdb/db/query/udf/core/transformer/ArithmeticNegationTransformer.java b/server/src/main/java/org/apache/iotdb/db/query/udf/core/transformer/ArithmeticNegationTransformer.java
index ca670f4..27093ee 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/udf/core/transformer/ArithmeticNegationTransformer.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/udf/core/transformer/ArithmeticNegationTransformer.java
@@ -44,22 +44,26 @@ public class ArithmeticNegationTransformer extends Transformer {
return false;
}
cachedTime = layerPointReader.currentTime();
- switch (layerPointReader.getDataType()) {
- case INT32:
- cachedInt = -layerPointReader.currentInt();
- break;
- case INT64:
- cachedLong = -layerPointReader.currentLong();
- break;
- case FLOAT:
- cachedFloat = -layerPointReader.currentFloat();
- break;
- case DOUBLE:
- cachedDouble = -layerPointReader.currentDouble();
- break;
- default:
- throw new QueryProcessException(
- "Unsupported data type: " + layerPointReader.getDataType().toString());
+ if (layerPointReader.isCurrentNull()) {
+ currentNull = true;
+ } else {
+ switch (layerPointReader.getDataType()) {
+ case INT32:
+ cachedInt = -layerPointReader.currentInt();
+ break;
+ case INT64:
+ cachedLong = -layerPointReader.currentLong();
+ break;
+ case FLOAT:
+ cachedFloat = -layerPointReader.currentFloat();
+ break;
+ case DOUBLE:
+ cachedDouble = -layerPointReader.currentDouble();
+ break;
+ default:
+ throw new QueryProcessException(
+ "Unsupported data type: " + layerPointReader.getDataType().toString());
+ }
}
layerPointReader.readyForNext();
return true;
diff --git a/server/src/main/java/org/apache/iotdb/db/query/udf/core/transformer/Transformer.java b/server/src/main/java/org/apache/iotdb/db/query/udf/core/transformer/Transformer.java
index 7f5d902..2583609 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/udf/core/transformer/Transformer.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/udf/core/transformer/Transformer.java
@@ -37,6 +37,7 @@ public abstract class Transformer implements LayerPointReader {
protected double cachedDouble;
protected boolean cachedBoolean;
protected Binary cachedBinary;
+ protected boolean currentNull;
protected Transformer() {
hasCachedValue = false;
@@ -50,11 +51,13 @@ public abstract class Transformer implements LayerPointReader {
return hasCachedValue;
}
+ /** if this method returns true, at least one of the cached field should be set */
protected abstract boolean cacheValue() throws QueryProcessException, IOException;
@Override
- public final void readyForNext() {
+ public void readyForNext() {
hasCachedValue = false;
+ currentNull = false;
}
@Override
@@ -91,4 +94,9 @@ public abstract class Transformer implements LayerPointReader {
public final Binary currentBinary() {
return cachedBinary;
}
+
+ @Override
+ public final boolean isCurrentNull() {
+ return currentNull;
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/query/udf/core/transformer/TransparentTransformer.java b/server/src/main/java/org/apache/iotdb/db/query/udf/core/transformer/TransparentTransformer.java
new file mode 100644
index 0000000..650565d
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/query/udf/core/transformer/TransparentTransformer.java
@@ -0,0 +1,93 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iotdb.db.query.udf.core.transformer;
+
+import org.apache.iotdb.db.exception.query.QueryProcessException;
+import org.apache.iotdb.db.query.udf.core.reader.LayerPointReader;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+
+import java.io.IOException;
+
+/**
+ * this is a special transformer which outputs data just as input without any modification.
+ *
+ * <p>i.e. it's just the function f(x) = x.
+ *
+ * <p>It's mainly used for a UDF with aggregation query as its parameters.
+ */
+public class TransparentTransformer extends Transformer {
+
+ private final LayerPointReader reader;
+
+ public TransparentTransformer(LayerPointReader reader) {
+ super();
+ this.reader = reader;
+ }
+
+ @Override
+ public boolean isConstantPointReader() {
+ return false;
+ }
+
+ @Override
+ public TSDataType getDataType() {
+ return reader.getDataType();
+ }
+
+ @Override
+ protected boolean cacheValue() throws QueryProcessException, IOException {
+ if (!reader.next()) {
+ return false;
+ }
+ if (reader.isCurrentNull()) {
+ currentNull = true;
+ } else {
+ switch (reader.getDataType()) {
+ case BOOLEAN:
+ cachedBoolean = reader.currentBoolean();
+ break;
+ case DOUBLE:
+ cachedDouble = reader.currentDouble();
+ break;
+ case FLOAT:
+ cachedFloat = reader.currentFloat();
+ break;
+ case INT32:
+ cachedInt = reader.currentInt();
+ break;
+ case INT64:
+ cachedLong = reader.currentLong();
+ break;
+ case TEXT:
+ cachedBinary = reader.currentBinary();
+ break;
+ default:
+ throw new QueryProcessException("unsupported data type: " + reader.getDataType());
+ }
+ }
+ cachedTime = reader.currentTime();
+ return true;
+ }
+
+ @Override
+ public void readyForNext() {
+ super.readyForNext();
+ reader.readyForNext();
+ }
+}
diff --git a/server/src/main/java/org/apache/iotdb/db/query/udf/core/transformer/UDFQueryRowTransformer.java b/server/src/main/java/org/apache/iotdb/db/query/udf/core/transformer/UDFQueryRowTransformer.java
index 7b0d124..2b9250c 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/udf/core/transformer/UDFQueryRowTransformer.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/udf/core/transformer/UDFQueryRowTransformer.java
@@ -39,7 +39,10 @@ public class UDFQueryRowTransformer extends UDFQueryTransformer {
if (!layerRowReader.next()) {
return false;
}
- executor.execute(layerRowReader.currentRow());
+ if (layerRowReader.isCurrentNull()) {
+ currentNull = true;
+ }
+ executor.execute(layerRowReader.currentRow(), currentNull);
layerRowReader.readyForNext();
return true;
}
diff --git a/server/src/main/java/org/apache/iotdb/db/query/udf/core/transformer/UDFQueryTransformer.java b/server/src/main/java/org/apache/iotdb/db/query/udf/core/transformer/UDFQueryTransformer.java
index ab2d8c3..f714a4e 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/udf/core/transformer/UDFQueryTransformer.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/udf/core/transformer/UDFQueryTransformer.java
@@ -59,9 +59,13 @@ public abstract class UDFQueryTransformer extends Transformer {
}
protected final boolean cacheValueFromUDFOutput() throws QueryProcessException, IOException {
- boolean hasNext = udfOutput.next();
- if (hasNext) {
- cachedTime = udfOutput.currentTime();
+ if (!udfOutput.next()) {
+ return false;
+ }
+ cachedTime = udfOutput.currentTime();
+ if (udfOutput.isCurrentNull()) {
+ currentNull = true;
+ } else {
switch (udfOutputDataType) {
case INT32:
cachedInt = udfOutput.currentInt();
@@ -84,9 +88,9 @@ public abstract class UDFQueryTransformer extends Transformer {
default:
throw new UnSupportedDataTypeException(udfOutputDataType.toString());
}
- udfOutput.readyForNext();
}
- return hasNext;
+ udfOutput.readyForNext();
+ return true;
}
protected abstract boolean executeUDFOnce() throws QueryProcessException, IOException;
diff --git a/server/src/main/java/org/apache/iotdb/db/query/udf/datastructure/row/ElasticSerializableRowRecordList.java b/server/src/main/java/org/apache/iotdb/db/query/udf/datastructure/row/ElasticSerializableRowRecordList.java
index 8484f74..cc42563 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/udf/datastructure/row/ElasticSerializableRowRecordList.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/udf/datastructure/row/ElasticSerializableRowRecordList.java
@@ -20,9 +20,11 @@
package org.apache.iotdb.db.query.udf.datastructure.row;
import org.apache.iotdb.db.exception.query.QueryProcessException;
+import org.apache.iotdb.db.query.udf.core.layer.InputRowUtils;
import org.apache.iotdb.db.query.udf.datastructure.Cache;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.utils.Binary;
+import org.apache.iotdb.tsfile.utils.BitMap;
import java.io.IOException;
import java.util.ArrayList;
@@ -43,6 +45,7 @@ public class ElasticSerializableRowRecordList {
protected LRUCache cache;
protected List<SerializableRowRecordList> rowRecordLists;
+ protected List<BitMap> bitMaps;
protected int size;
protected int evictionUpperBound;
@@ -78,6 +81,7 @@ public class ElasticSerializableRowRecordList {
cache = new ElasticSerializableRowRecordList.LRUCache(numCacheBlock);
rowRecordLists = new ArrayList<>();
+ bitMaps = new ArrayList<>();
size = 0;
evictionUpperBound = 0;
@@ -115,6 +119,7 @@ public class ElasticSerializableRowRecordList {
cache = new ElasticSerializableRowRecordList.LRUCache(numCacheBlock);
rowRecordLists = new ArrayList<>();
+ bitMaps = new ArrayList<>();
size = 0;
evictionUpperBound = 0;
@@ -141,9 +146,28 @@ public class ElasticSerializableRowRecordList {
.getRowRecord(index % internalRowRecordListCapacity);
}
+ /** true if all fields except the timestamp in the current row are null */
+ public boolean fieldsAllNull(int index) {
+ return bitMaps
+ .get(index / internalRowRecordListCapacity)
+ .isMarked(index % internalRowRecordListCapacity);
+ }
+
public void put(Object[] rowRecord) throws IOException, QueryProcessException {
+ put(rowRecord, InputRowUtils.isAllNull(rowRecord));
+ }
+
+ /**
+ * Put the row in the list with an all-fields-null marker, this method is faster than calling put
+ * directly
+ */
+ private void put(Object[] rowRecord, boolean fieldsAllNull)
+ throws IOException, QueryProcessException {
checkExpansion();
cache.get(size / internalRowRecordListCapacity).put(rowRecord);
+ if (fieldsAllNull) {
+ bitMaps.get(size / internalRowRecordListCapacity).mark(size % internalRowRecordListCapacity);
+ }
++size;
if (!disableMemoryControl) {
@@ -168,6 +192,7 @@ public class ElasticSerializableRowRecordList {
rowRecordLists.add(
SerializableRowRecordList.newSerializableRowRecordList(
queryId, dataTypes, internalRowRecordListCapacity));
+ bitMaps.add(new BitMap(internalRowRecordListCapacity));
}
}
@@ -223,6 +248,7 @@ public class ElasticSerializableRowRecordList {
int internalListEvictionUpperBound = evictionUpperBound / newInternalRowRecordListCapacity;
for (int i = 0; i < internalListEvictionUpperBound; ++i) {
newElasticSerializableRowRecordList.rowRecordLists.add(null);
+ newElasticSerializableRowRecordList.bitMaps.add(null);
}
newElasticSerializableRowRecordList.size =
internalListEvictionUpperBound * newInternalRowRecordListCapacity;
@@ -230,12 +256,13 @@ public class ElasticSerializableRowRecordList {
newElasticSerializableRowRecordList.put(null);
}
for (int i = evictionUpperBound; i < size; ++i) {
- newElasticSerializableRowRecordList.put(getRowRecord(i));
+ newElasticSerializableRowRecordList.put(getRowRecord(i), fieldsAllNull(i));
}
internalRowRecordListCapacity = newInternalRowRecordListCapacity;
cache = newElasticSerializableRowRecordList.cache;
rowRecordLists = newElasticSerializableRowRecordList.rowRecordLists;
+ bitMaps = newElasticSerializableRowRecordList.bitMaps;
byteArrayLengthForMemoryControl = newByteArrayLengthForMemoryControl;
totalByteArrayLengthLimit =
@@ -262,6 +289,7 @@ public class ElasticSerializableRowRecordList {
int lastIndex = removeLast();
if (lastIndex < evictionUpperBound / internalRowRecordListCapacity) {
rowRecordLists.set(lastIndex, null);
+ bitMaps.set(lastIndex, null);
} else {
rowRecordLists.get(lastIndex).serialize();
}
diff --git a/server/src/main/java/org/apache/iotdb/db/query/udf/datastructure/row/SerializableRowRecordList.java b/server/src/main/java/org/apache/iotdb/db/query/udf/datastructure/row/SerializableRowRecordList.java
index 8a76056..eab6af6 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/udf/datastructure/row/SerializableRowRecordList.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/udf/datastructure/row/SerializableRowRecordList.java
@@ -83,7 +83,8 @@ public class SerializableRowRecordList implements SerializableList {
}
}
- int size = (int) (memoryLimitInMB * MB / 2 / rowLength);
+ // 1 extra bit for null fields mark in bitMap
+ int size = (int) (memoryLimitInMB * MB / 2 / (rowLength + ReadWriteIOUtils.BIT_LEN));
if (size <= 0) {
throw new QueryProcessException("Memory is not enough for current query.");
}
diff --git a/server/src/main/java/org/apache/iotdb/db/query/udf/datastructure/tv/ElasticSerializableBinaryTVList.java b/server/src/main/java/org/apache/iotdb/db/query/udf/datastructure/tv/ElasticSerializableBinaryTVList.java
index ad2d9e9..0e481b7 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/udf/datastructure/tv/ElasticSerializableBinaryTVList.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/udf/datastructure/tv/ElasticSerializableBinaryTVList.java
@@ -112,6 +112,7 @@ public class ElasticSerializableBinaryTVList extends ElasticSerializableTVList {
int internalListEvictionUpperBound = evictionUpperBound / newInternalTVListCapacity;
for (int i = 0; i < internalListEvictionUpperBound; ++i) {
newElasticSerializableTVList.tvLists.add(null);
+ newElasticSerializableTVList.bitMaps.add(null);
}
newElasticSerializableTVList.size = internalListEvictionUpperBound * newInternalTVListCapacity;
Binary empty = Binary.valueOf("");
@@ -119,12 +120,17 @@ public class ElasticSerializableBinaryTVList extends ElasticSerializableTVList {
newElasticSerializableTVList.putBinary(i, empty);
}
for (int i = evictionUpperBound; i < size; ++i) {
- newElasticSerializableTVList.putBinary(getTime(i), getBinary(i));
+ if (isNull(i)) {
+ newElasticSerializableTVList.putNull(getTime(i));
+ } else {
+ newElasticSerializableTVList.putBinary(getTime(i), getBinary(i));
+ }
}
internalTVListCapacity = newInternalTVListCapacity;
cache = newElasticSerializableTVList.cache;
tvLists = newElasticSerializableTVList.tvLists;
+ bitMaps = newElasticSerializableTVList.bitMaps;
byteArrayLengthForMemoryControl = newByteArrayLengthForMemoryControl;
totalByteArrayLengthLimit = (long) size * byteArrayLengthForMemoryControl;
diff --git a/server/src/main/java/org/apache/iotdb/db/query/udf/datastructure/tv/ElasticSerializableTVList.java b/server/src/main/java/org/apache/iotdb/db/query/udf/datastructure/tv/ElasticSerializableTVList.java
index 38ac4f5..adf9abc 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/udf/datastructure/tv/ElasticSerializableTVList.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/udf/datastructure/tv/ElasticSerializableTVList.java
@@ -27,6 +27,7 @@ import org.apache.iotdb.tsfile.exception.write.UnSupportedDataTypeException;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.read.common.BatchData;
import org.apache.iotdb.tsfile.utils.Binary;
+import org.apache.iotdb.tsfile.utils.BitMap;
import java.io.IOException;
import java.util.ArrayList;
@@ -51,6 +52,12 @@ public class ElasticSerializableTVList implements PointCollector {
protected LRUCache cache;
protected List<SerializableTVList> tvLists;
+ /**
+ * the bitmap used to indicate whether one value is null in the tvLists. The size of bitMap is the
+ * same as tvLists and the length of whole bits is the same as tvLists' length.
+ */
+ protected List<BitMap> bitMaps;
+
protected int size;
protected int evictionUpperBound;
@@ -69,6 +76,7 @@ public class ElasticSerializableTVList implements PointCollector {
this.cacheSize = cacheSize;
cache = new LRUCache(cacheSize);
+ bitMaps = new ArrayList<>();
tvLists = new ArrayList<>();
size = 0;
evictionUpperBound = 0;
@@ -87,6 +95,7 @@ public class ElasticSerializableTVList implements PointCollector {
this.cacheSize = cacheSize;
cache = new LRUCache(cacheSize);
+ bitMaps = new ArrayList<>();
tvLists = new ArrayList<>();
size = 0;
evictionUpperBound = 0;
@@ -100,6 +109,10 @@ public class ElasticSerializableTVList implements PointCollector {
return size;
}
+ public boolean isNull(int index) throws IOException {
+ return bitMaps.get(index / internalTVListCapacity).isMarked(index % internalTVListCapacity);
+ }
+
public long getTime(int index) throws IOException {
return cache.get(index / internalTVListCapacity).getTimeByIndex(index % internalTVListCapacity);
}
@@ -218,9 +231,37 @@ public class ElasticSerializableTVList implements PointCollector {
++size;
}
+ public void putNull(long timestamp) throws IOException, QueryProcessException {
+ switch (dataType) {
+ case INT32:
+ putInt(timestamp, 0);
+ break;
+ case INT64:
+ putLong(timestamp, 0L);
+ break;
+ case FLOAT:
+ putFloat(timestamp, 0.0F);
+ break;
+ case DOUBLE:
+ putDouble(timestamp, 0.0D);
+ break;
+ case BOOLEAN:
+ putBoolean(timestamp, false);
+ break;
+ case TEXT:
+ putBinary(timestamp, Binary.EMPTY_VALUE);
+ break;
+ default:
+ throw new UnSupportedDataTypeException(
+ String.format("Data type %s is not supported.", dataType));
+ }
+ bitMaps.get((size - 1) / internalTVListCapacity).mark((size - 1) % internalTVListCapacity);
+ }
+
private void checkExpansion() {
if (size % internalTVListCapacity == 0) {
tvLists.add(SerializableTVList.newSerializableTVList(dataType, queryId));
+ bitMaps.add(new BitMap(internalTVListCapacity));
}
}
@@ -288,6 +329,11 @@ public class ElasticSerializableTVList implements PointCollector {
public Binary currentBinary() throws IOException {
return getBinary(currentPointIndex);
}
+
+ @Override
+ public boolean isCurrentNull() throws IOException {
+ return isNull(currentPointIndex);
+ }
};
}
@@ -311,6 +357,7 @@ public class ElasticSerializableTVList implements PointCollector {
int lastIndex = removeLast();
if (lastIndex < evictionUpperBound / internalTVListCapacity) {
tvLists.set(lastIndex, null);
+ bitMaps.set(lastIndex, null);
} else {
tvLists.get(lastIndex).serialize();
}
diff --git a/server/src/main/java/org/apache/iotdb/db/query/udf/datastructure/tv/SerializableBinaryTVList.java b/server/src/main/java/org/apache/iotdb/db/query/udf/datastructure/tv/SerializableBinaryTVList.java
index c991dfe..7553d47 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/udf/datastructure/tv/SerializableBinaryTVList.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/udf/datastructure/tv/SerializableBinaryTVList.java
@@ -43,11 +43,12 @@ public class SerializableBinaryTVList extends SerializableTVList {
* calculateSingleBinaryTVPairMemory(byteArrayLength)));
}
- protected static int calculateSingleBinaryTVPairMemory(int byteArrayLength) {
+ protected static float calculateSingleBinaryTVPairMemory(int byteArrayLength) {
return ReadWriteIOUtils.LONG_LEN // time
+ MIN_OBJECT_HEADER_SIZE // value: header length of Binary
+ MIN_ARRAY_HEADER_SIZE // value: header length of values in Binary
- + byteArrayLength; // value: length of values array in Binary
+ + byteArrayLength // value: length of values array in Binary
+ + ReadWriteIOUtils.BIT_LEN; // extra bit(1/8 Byte): whether the Binary is null
}
protected SerializableBinaryTVList(SerializationRecorder serializationRecorder) {
diff --git a/server/src/main/java/org/apache/iotdb/db/query/udf/datastructure/tv/SerializableBooleanTVList.java b/server/src/main/java/org/apache/iotdb/db/query/udf/datastructure/tv/SerializableBooleanTVList.java
index 48ec1d2..f2ed81c 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/udf/datastructure/tv/SerializableBooleanTVList.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/udf/datastructure/tv/SerializableBooleanTVList.java
@@ -33,10 +33,13 @@ public class SerializableBooleanTVList extends SerializableTVList {
protected static int calculateCapacity(float memoryLimitInMB) {
float memoryLimitInB = memoryLimitInMB * MB / 2;
+ // One time-value pair with 1 extra bit(1/8 Byte) in bitMap
return TSFileConfig.ARRAY_CAPACITY_THRESHOLD
* (int)
(memoryLimitInB
- / ((ReadWriteIOUtils.LONG_LEN + ReadWriteIOUtils.BOOLEAN_LEN)
+ / ((ReadWriteIOUtils.LONG_LEN
+ + ReadWriteIOUtils.BOOLEAN_LEN
+ + ReadWriteIOUtils.BIT_LEN)
* TSFileConfig.ARRAY_CAPACITY_THRESHOLD));
}
diff --git a/server/src/main/java/org/apache/iotdb/db/query/udf/datastructure/tv/SerializableDoubleTVList.java b/server/src/main/java/org/apache/iotdb/db/query/udf/datastructure/tv/SerializableDoubleTVList.java
index 6f55333..3db3326 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/udf/datastructure/tv/SerializableDoubleTVList.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/udf/datastructure/tv/SerializableDoubleTVList.java
@@ -33,10 +33,13 @@ public class SerializableDoubleTVList extends SerializableTVList {
protected static int calculateCapacity(float memoryLimitInMB) {
float memoryLimitInB = memoryLimitInMB * MB / 2;
+ // One time-value pair with 1 extra bit(1/8 Byte) in bitMap
return TSFileConfig.ARRAY_CAPACITY_THRESHOLD
* (int)
(memoryLimitInB
- / ((ReadWriteIOUtils.LONG_LEN + ReadWriteIOUtils.DOUBLE_LEN)
+ / ((ReadWriteIOUtils.LONG_LEN
+ + ReadWriteIOUtils.DOUBLE_LEN
+ + ReadWriteIOUtils.BIT_LEN)
* TSFileConfig.ARRAY_CAPACITY_THRESHOLD));
}
diff --git a/server/src/main/java/org/apache/iotdb/db/query/udf/datastructure/tv/SerializableFloatTVList.java b/server/src/main/java/org/apache/iotdb/db/query/udf/datastructure/tv/SerializableFloatTVList.java
index ebfee41..b7ec608 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/udf/datastructure/tv/SerializableFloatTVList.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/udf/datastructure/tv/SerializableFloatTVList.java
@@ -33,10 +33,13 @@ public class SerializableFloatTVList extends SerializableTVList {
protected static int calculateCapacity(float memoryLimitInMB) {
float memoryLimitInB = memoryLimitInMB * MB / 2;
+ // One time-value pair with 1 extra bit(1/8 Byte) in bitMap
return TSFileConfig.ARRAY_CAPACITY_THRESHOLD
* (int)
(memoryLimitInB
- / ((ReadWriteIOUtils.LONG_LEN + ReadWriteIOUtils.FLOAT_LEN)
+ / ((ReadWriteIOUtils.LONG_LEN
+ + ReadWriteIOUtils.FLOAT_LEN
+ + ReadWriteIOUtils.BIT_LEN)
* TSFileConfig.ARRAY_CAPACITY_THRESHOLD));
}
diff --git a/server/src/main/java/org/apache/iotdb/db/query/udf/datastructure/tv/SerializableIntTVList.java b/server/src/main/java/org/apache/iotdb/db/query/udf/datastructure/tv/SerializableIntTVList.java
index 96445f4..28454dd 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/udf/datastructure/tv/SerializableIntTVList.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/udf/datastructure/tv/SerializableIntTVList.java
@@ -33,10 +33,11 @@ public class SerializableIntTVList extends SerializableTVList {
protected static int calculateCapacity(float memoryLimitInMB) {
float memoryLimitInB = memoryLimitInMB * MB / 2;
+ // One time-value pair with 1 extra bit(1/8 Byte) in bitMap
return TSFileConfig.ARRAY_CAPACITY_THRESHOLD
* (int)
(memoryLimitInB
- / ((ReadWriteIOUtils.LONG_LEN + ReadWriteIOUtils.INT_LEN)
+ / ((ReadWriteIOUtils.LONG_LEN + ReadWriteIOUtils.INT_LEN + ReadWriteIOUtils.BIT_LEN)
* TSFileConfig.ARRAY_CAPACITY_THRESHOLD));
}
diff --git a/server/src/main/java/org/apache/iotdb/db/query/udf/datastructure/tv/SerializableLongTVList.java b/server/src/main/java/org/apache/iotdb/db/query/udf/datastructure/tv/SerializableLongTVList.java
index 4bfd3b6..c4cc3ec 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/udf/datastructure/tv/SerializableLongTVList.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/udf/datastructure/tv/SerializableLongTVList.java
@@ -33,10 +33,13 @@ public class SerializableLongTVList extends SerializableTVList {
protected static int calculateCapacity(float memoryLimitInMB) {
float memoryLimitInB = memoryLimitInMB * MB / 2;
+ // One time-value pair with 1 extra bit(1/8 Byte) in bitMap
return TSFileConfig.ARRAY_CAPACITY_THRESHOLD
* (int)
(memoryLimitInB
- / ((ReadWriteIOUtils.LONG_LEN + ReadWriteIOUtils.LONG_LEN)
+ / ((ReadWriteIOUtils.LONG_LEN
+ + ReadWriteIOUtils.LONG_LEN
+ + ReadWriteIOUtils.BIT_LEN)
* TSFileConfig.ARRAY_CAPACITY_THRESHOLD));
}
diff --git a/server/src/main/java/org/apache/iotdb/db/service/thrift/impl/TSServiceImpl.java b/server/src/main/java/org/apache/iotdb/db/service/thrift/impl/TSServiceImpl.java
index 0683ac7..3fda7c4 100644
--- a/server/src/main/java/org/apache/iotdb/db/service/thrift/impl/TSServiceImpl.java
+++ b/server/src/main/java/org/apache/iotdb/db/service/thrift/impl/TSServiceImpl.java
@@ -38,8 +38,34 @@ import org.apache.iotdb.db.metadata.template.TemplateQueryType;
import org.apache.iotdb.db.qp.constant.SQLConstant;
import org.apache.iotdb.db.qp.logical.Operator.OperatorType;
import org.apache.iotdb.db.qp.physical.PhysicalPlan;
-import org.apache.iotdb.db.qp.physical.crud.*;
-import org.apache.iotdb.db.qp.physical.sys.*;
+import org.apache.iotdb.db.qp.physical.crud.AggregationPlan;
+import org.apache.iotdb.db.qp.physical.crud.AlignByDevicePlan;
+import org.apache.iotdb.db.qp.physical.crud.DeletePlan;
+import org.apache.iotdb.db.qp.physical.crud.InsertMultiTabletPlan;
+import org.apache.iotdb.db.qp.physical.crud.InsertRowPlan;
+import org.apache.iotdb.db.qp.physical.crud.InsertRowsOfOneDevicePlan;
+import org.apache.iotdb.db.qp.physical.crud.InsertRowsPlan;
+import org.apache.iotdb.db.qp.physical.crud.InsertTabletPlan;
+import org.apache.iotdb.db.qp.physical.crud.LastQueryPlan;
+import org.apache.iotdb.db.qp.physical.crud.MeasurementInfo;
+import org.apache.iotdb.db.qp.physical.crud.QueryPlan;
+import org.apache.iotdb.db.qp.physical.crud.SelectIntoPlan;
+import org.apache.iotdb.db.qp.physical.crud.UDAFPlan;
+import org.apache.iotdb.db.qp.physical.crud.UDFPlan;
+import org.apache.iotdb.db.qp.physical.sys.AppendTemplatePlan;
+import org.apache.iotdb.db.qp.physical.sys.AuthorPlan;
+import org.apache.iotdb.db.qp.physical.sys.CreateAlignedTimeSeriesPlan;
+import org.apache.iotdb.db.qp.physical.sys.CreateMultiTimeSeriesPlan;
+import org.apache.iotdb.db.qp.physical.sys.CreateTemplatePlan;
+import org.apache.iotdb.db.qp.physical.sys.CreateTimeSeriesPlan;
+import org.apache.iotdb.db.qp.physical.sys.DeleteStorageGroupPlan;
+import org.apache.iotdb.db.qp.physical.sys.DeleteTimeSeriesPlan;
+import org.apache.iotdb.db.qp.physical.sys.PruneTemplatePlan;
+import org.apache.iotdb.db.qp.physical.sys.SetStorageGroupPlan;
+import org.apache.iotdb.db.qp.physical.sys.SetTemplatePlan;
+import org.apache.iotdb.db.qp.physical.sys.ShowPlan;
+import org.apache.iotdb.db.qp.physical.sys.ShowQueryProcesslistPlan;
+import org.apache.iotdb.db.qp.physical.sys.UnsetTemplatePlan;
import org.apache.iotdb.db.query.aggregation.AggregateResult;
import org.apache.iotdb.db.query.context.QueryContext;
import org.apache.iotdb.db.query.control.tracing.TracingConstant;
@@ -60,7 +86,47 @@ import org.apache.iotdb.metrics.MetricService;
import org.apache.iotdb.rpc.RedirectException;
import org.apache.iotdb.rpc.RpcUtils;
import org.apache.iotdb.rpc.TSStatusCode;
-import org.apache.iotdb.service.rpc.thrift.*;
+import org.apache.iotdb.service.rpc.thrift.EndPoint;
+import org.apache.iotdb.service.rpc.thrift.ServerProperties;
+import org.apache.iotdb.service.rpc.thrift.TSAppendSchemaTemplateReq;
+import org.apache.iotdb.service.rpc.thrift.TSCancelOperationReq;
+import org.apache.iotdb.service.rpc.thrift.TSCloseOperationReq;
+import org.apache.iotdb.service.rpc.thrift.TSCloseSessionReq;
+import org.apache.iotdb.service.rpc.thrift.TSCreateAlignedTimeseriesReq;
+import org.apache.iotdb.service.rpc.thrift.TSCreateMultiTimeseriesReq;
+import org.apache.iotdb.service.rpc.thrift.TSCreateSchemaTemplateReq;
+import org.apache.iotdb.service.rpc.thrift.TSCreateTimeseriesReq;
+import org.apache.iotdb.service.rpc.thrift.TSDeleteDataReq;
+import org.apache.iotdb.service.rpc.thrift.TSExecuteBatchStatementReq;
+import org.apache.iotdb.service.rpc.thrift.TSExecuteStatementReq;
+import org.apache.iotdb.service.rpc.thrift.TSExecuteStatementResp;
+import org.apache.iotdb.service.rpc.thrift.TSFetchMetadataReq;
+import org.apache.iotdb.service.rpc.thrift.TSFetchMetadataResp;
+import org.apache.iotdb.service.rpc.thrift.TSFetchResultsReq;
+import org.apache.iotdb.service.rpc.thrift.TSFetchResultsResp;
+import org.apache.iotdb.service.rpc.thrift.TSGetTimeZoneResp;
+import org.apache.iotdb.service.rpc.thrift.TSIService;
+import org.apache.iotdb.service.rpc.thrift.TSInsertRecordReq;
+import org.apache.iotdb.service.rpc.thrift.TSInsertRecordsOfOneDeviceReq;
+import org.apache.iotdb.service.rpc.thrift.TSInsertRecordsReq;
+import org.apache.iotdb.service.rpc.thrift.TSInsertStringRecordReq;
+import org.apache.iotdb.service.rpc.thrift.TSInsertStringRecordsReq;
+import org.apache.iotdb.service.rpc.thrift.TSInsertTabletReq;
+import org.apache.iotdb.service.rpc.thrift.TSInsertTabletsReq;
+import org.apache.iotdb.service.rpc.thrift.TSLastDataQueryReq;
+import org.apache.iotdb.service.rpc.thrift.TSOpenSessionReq;
+import org.apache.iotdb.service.rpc.thrift.TSOpenSessionResp;
+import org.apache.iotdb.service.rpc.thrift.TSPruneSchemaTemplateReq;
+import org.apache.iotdb.service.rpc.thrift.TSQueryDataSet;
+import org.apache.iotdb.service.rpc.thrift.TSQueryNonAlignDataSet;
+import org.apache.iotdb.service.rpc.thrift.TSQueryTemplateReq;
+import org.apache.iotdb.service.rpc.thrift.TSQueryTemplateResp;
+import org.apache.iotdb.service.rpc.thrift.TSRawDataQueryReq;
+import org.apache.iotdb.service.rpc.thrift.TSSetSchemaTemplateReq;
+import org.apache.iotdb.service.rpc.thrift.TSSetTimeZoneReq;
+import org.apache.iotdb.service.rpc.thrift.TSStatus;
+import org.apache.iotdb.service.rpc.thrift.TSTracingInfo;
+import org.apache.iotdb.service.rpc.thrift.TSUnsetSchemaTemplateReq;
import org.apache.iotdb.tsfile.exception.filter.QueryFilterOptimizationException;
import org.apache.iotdb.tsfile.exception.write.UnSupportedDataTypeException;
import org.apache.iotdb.tsfile.file.metadata.enums.CompressionType;
@@ -78,11 +144,22 @@ import java.io.IOException;
import java.nio.ByteBuffer;
import java.sql.SQLException;
import java.time.ZoneId;
-import java.util.*;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.BitSet;
+import java.util.Collections;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Map;
import java.util.Map.Entry;
+import java.util.Set;
import java.util.stream.Collectors;
-import static org.apache.iotdb.db.utils.ErrorHandlingUtils.*;
+import static org.apache.iotdb.db.utils.ErrorHandlingUtils.onIoTDBException;
+import static org.apache.iotdb.db.utils.ErrorHandlingUtils.onNPEOrUnexpectedException;
+import static org.apache.iotdb.db.utils.ErrorHandlingUtils.onNonQueryException;
+import static org.apache.iotdb.db.utils.ErrorHandlingUtils.onQueryException;
+import static org.apache.iotdb.db.utils.ErrorHandlingUtils.tryCatchQueryException;
/** Thrift RPC implementation at server side. */
public class TSServiceImpl extends BasicServiceProvider implements TSIService.Iface {
@@ -604,7 +681,10 @@ public class TSServiceImpl extends BasicServiceProvider implements TSIService.If
}
resp.setOperationType(plan.getOperatorType().toString());
- if (plan.getOperatorType() == OperatorType.AGGREGATION) {
+ if (plan.getOperatorType() == OperatorType.AGGREGATION
+ || (plan instanceof UDAFPlan
+ && ((UDAFPlan) plan).getInnerAggregationPlan().getOperatorType()
+ == OperatorType.AGGREGATION)) {
resp.setIgnoreTimeStamp(true);
} else if (plan instanceof ShowQueryProcesslistPlan) {
resp.setIgnoreTimeStamp(false);
@@ -766,6 +846,7 @@ public class TSServiceImpl extends BasicServiceProvider implements TSIService.If
}
seriesTypes = SchemaUtils.getSeriesTypesByPaths(paths, aggregations);
break;
+ case UDAF:
case UDTF:
seriesTypes = new ArrayList<>();
for (int i = 0; i < paths.size(); i++) {
diff --git a/server/src/test/java/org/apache/iotdb/db/query/udf/datastructure/ElasticSerializableRowRecordListTest.java b/server/src/test/java/org/apache/iotdb/db/query/udf/datastructure/ElasticSerializableRowRecordListTest.java
index 05d5996..7704a1e 100644
--- a/server/src/test/java/org/apache/iotdb/db/query/udf/datastructure/ElasticSerializableRowRecordListTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/query/udf/datastructure/ElasticSerializableRowRecordListTest.java
@@ -32,6 +32,8 @@ import java.io.IOException;
import java.util.Random;
import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
public class ElasticSerializableRowRecordListTest extends SerializableListTest {
@@ -85,28 +87,31 @@ public class ElasticSerializableRowRecordListTest extends SerializableListTest {
for (int i = 0; i < ITERATION_TIMES; ++i) {
Object[] rowRecord = new Object[DATA_TYPES.length + 1];
rowRecord[DATA_TYPES.length] = (long) i;
- for (int j = 0; j < DATA_TYPES.length; ++j) {
- switch (DATA_TYPES[j]) {
- case INT32:
- rowRecord[j] = i;
- break;
- case INT64:
- rowRecord[j] = (long) i;
- break;
- case FLOAT:
- rowRecord[j] = (float) i;
- break;
- case DOUBLE:
- rowRecord[j] = (double) i;
- break;
- case BOOLEAN:
- rowRecord[j] = i % 2 == 0;
- break;
- case TEXT:
- rowRecord[j] = Binary.valueOf(String.valueOf(i));
- break;
+ if (i % 7 != 0) {
+ for (int j = 0; j < DATA_TYPES.length; ++j) {
+ switch (DATA_TYPES[j]) {
+ case INT32:
+ rowRecord[j] = i;
+ break;
+ case INT64:
+ rowRecord[j] = (long) i;
+ break;
+ case FLOAT:
+ rowRecord[j] = (float) i;
+ break;
+ case DOUBLE:
+ rowRecord[j] = (double) i;
+ break;
+ case BOOLEAN:
+ rowRecord[j] = i % 2 == 0;
+ break;
+ case TEXT:
+ rowRecord[j] = Binary.valueOf(String.valueOf(i));
+ break;
+ }
}
}
+
rowRecordList.put(rowRecord);
}
} catch (IOException | QueryProcessException e) {
@@ -144,7 +149,12 @@ public class ElasticSerializableRowRecordListTest extends SerializableListTest {
private void testOrderedAccessByIndex() {
try {
for (int i = 0; i < ITERATION_TIMES; ++i) {
- testRowRecord(rowRecordList.getRowRecord(i), i);
+ if (i % 7 == 0) {
+ assertTrue(rowRecordList.fieldsAllNull(i));
+ } else {
+ assertFalse(rowRecordList.fieldsAllNull(i));
+ testRowRecord(rowRecordList.getRowRecord(i), i);
+ }
}
} catch (IOException e) {
fail(e.toString());
@@ -160,34 +170,79 @@ public class ElasticSerializableRowRecordListTest extends SerializableListTest {
Random random = new Random();
try {
- for (int i = 0; i < ITERATION_TIMES; ++i) {
- rowRecordList.put(
- generateRowRecord(i, byteLengthMin + random.nextInt(byteLengthMax - byteLengthMin)));
+ for (int i = 0; i < ITERATION_TIMES; i++) {
+ if (i % 7 == 0) {
+ rowRecordList.put(generateRowRecordWithAllNullFields(i));
+ } else {
+ rowRecordList.put(
+ generateRowRecord(i, byteLengthMin + random.nextInt(byteLengthMax - byteLengthMin)));
+ }
}
rowRecordList.setEvictionUpperBound(rowRecordList.size());
+ for (int i = 0; i < ITERATION_TIMES; i++) {
+ if (i % 7 == 0) {
+ assertTrue(rowRecordList.fieldsAllNull(i));
+ } else {
+ assertFalse(rowRecordList.fieldsAllNull(i));
+ }
+ }
byteLengthMin = SerializableList.INITIAL_BYTE_ARRAY_LENGTH_FOR_MEMORY_CONTROL * 16;
byteLengthMax = SerializableList.INITIAL_BYTE_ARRAY_LENGTH_FOR_MEMORY_CONTROL * 32;
- for (int i = 0; i < ITERATION_TIMES; ++i) {
- rowRecordList.put(
- generateRowRecord(i, byteLengthMin + random.nextInt(byteLengthMax - byteLengthMin)));
+ for (int i = 0; i < ITERATION_TIMES; i++) {
+ if (i % 7 == 0) {
+ rowRecordList.put(generateRowRecordWithAllNullFields(i));
+ } else {
+ rowRecordList.put(
+ generateRowRecord(i, byteLengthMin + random.nextInt(byteLengthMax - byteLengthMin)));
+ }
}
rowRecordList.setEvictionUpperBound(rowRecordList.size());
+ for (int i = 0; i < ITERATION_TIMES; i++) {
+ if (i % 7 == 0) {
+ assertTrue(rowRecordList.fieldsAllNull(i + ITERATION_TIMES));
+ } else {
+ assertFalse(rowRecordList.fieldsAllNull(i + ITERATION_TIMES));
+ }
+ }
byteLengthMin = SerializableList.INITIAL_BYTE_ARRAY_LENGTH_FOR_MEMORY_CONTROL * 256;
byteLengthMax = SerializableList.INITIAL_BYTE_ARRAY_LENGTH_FOR_MEMORY_CONTROL * 512;
- for (int i = 0; i < ITERATION_TIMES; ++i) {
- rowRecordList.put(
- generateRowRecord(i, byteLengthMin + random.nextInt(byteLengthMax - byteLengthMin)));
+ for (int i = 0; i < ITERATION_TIMES; i++) {
+ if (i % 7 == 0) {
+ rowRecordList.put(generateRowRecordWithAllNullFields(i));
+ } else {
+ rowRecordList.put(
+ generateRowRecord(i, byteLengthMin + random.nextInt(byteLengthMax - byteLengthMin)));
+ }
}
rowRecordList.setEvictionUpperBound(rowRecordList.size());
+ for (int i = 0; i < ITERATION_TIMES; i++) {
+ if (i % 7 == 0) {
+ assertTrue(rowRecordList.fieldsAllNull(i + 2 * ITERATION_TIMES));
+ } else {
+ assertFalse(rowRecordList.fieldsAllNull(i + 2 * ITERATION_TIMES));
+ }
+ }
- for (int i = 0; i < 2 * ITERATION_TIMES; ++i) {
- rowRecordList.put(
- generateRowRecord(i, byteLengthMin + random.nextInt(byteLengthMax - byteLengthMin)));
+ for (int i = 0; i < 2 * ITERATION_TIMES; i++) {
+ if (i % 7 == 0) {
+ rowRecordList.put(generateRowRecordWithAllNullFields(i));
+ } else {
+ rowRecordList.put(
+ generateRowRecord(i, byteLengthMin + random.nextInt(byteLengthMax - byteLengthMin)));
+ }
rowRecordList.setEvictionUpperBound(rowRecordList.size());
}
+ for (int i = 0; i < ITERATION_TIMES; i++) {
+ if (i % 7 == 0) {
+ assertTrue(rowRecordList.fieldsAllNull(i + 3 * ITERATION_TIMES));
+ } else {
+ assertFalse(rowRecordList.fieldsAllNull(i + 3 * ITERATION_TIMES));
+ }
+ }
+
assertEquals(ITERATION_TIMES * 5, rowRecordList.size());
} catch (QueryProcessException | IOException e) {
e.printStackTrace();
@@ -223,6 +278,12 @@ public class ElasticSerializableRowRecordListTest extends SerializableListTest {
return rowRecord;
}
+ private Object[] generateRowRecordWithAllNullFields(int time) {
+ Object[] rowRecord = new Object[DATA_TYPES.length + 1];
+ rowRecord[DATA_TYPES.length] = (long) time;
+ return rowRecord;
+ }
+
private String generateRandomString(int length) {
StringBuilder stringBuilder = new StringBuilder();
for (int i = 0; i < length; ++i) {
diff --git a/server/src/test/java/org/apache/iotdb/db/query/udf/datastructure/ElasticSerializableTVListTest.java b/server/src/test/java/org/apache/iotdb/db/query/udf/datastructure/ElasticSerializableTVListTest.java
index 280e340..865c270 100644
--- a/server/src/test/java/org/apache/iotdb/db/query/udf/datastructure/ElasticSerializableTVListTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/query/udf/datastructure/ElasticSerializableTVListTest.java
@@ -33,6 +33,7 @@ import java.io.IOException;
import java.util.Random;
import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
@@ -106,32 +107,56 @@ public class ElasticSerializableTVListTest extends SerializableListTest {
switch (dataType) {
case INT32:
for (int i = 0; i < ITERATION_TIMES; ++i) {
- tvList.putInt(i, i);
+ if (i % 7 == 0) {
+ tvList.putNull(i);
+ } else {
+ tvList.putInt(i, i);
+ }
}
break;
case INT64:
for (int i = 0; i < ITERATION_TIMES; ++i) {
- tvList.putLong(i, i);
+ if (i % 7 == 0) {
+ tvList.putNull(i);
+ } else {
+ tvList.putLong(i, i);
+ }
}
break;
case FLOAT:
for (int i = 0; i < ITERATION_TIMES; ++i) {
- tvList.putFloat(i, i);
+ if (i % 7 == 0) {
+ tvList.putNull(i);
+ } else {
+ tvList.putFloat(i, i);
+ }
}
break;
case DOUBLE:
for (int i = 0; i < ITERATION_TIMES; ++i) {
- tvList.putDouble(i, i);
+ if (i % 7 == 0) {
+ tvList.putNull(i);
+ } else {
+ tvList.putDouble(i, i);
+ }
}
break;
case BOOLEAN:
for (int i = 0; i < ITERATION_TIMES; ++i) {
- tvList.putBoolean(i, i % 2 == 0);
+ if (i % 7 == 0) {
+ tvList.putNull(i);
+ } else {
+ tvList.putBoolean(i, i % 2 == 0);
+ }
}
break;
case TEXT:
for (int i = 0; i < ITERATION_TIMES; ++i) {
- tvList.putBinary(i, Binary.valueOf(String.valueOf(i)));
+ if (i % 7 == 0) {
+ tvList.putNull(i);
+ } else {
+ tvList.putBinary(i, Binary.valueOf(String.valueOf(i)));
+ }
}
break;
}
@@ -147,37 +172,67 @@ public class ElasticSerializableTVListTest extends SerializableListTest {
case INT32:
for (int i = 0; i < ITERATION_TIMES; ++i) {
assertEquals(i, tvList.getTime(i));
- assertEquals(i, tvList.getInt(i));
+ if (i % 7 == 0) {
+ assertTrue(tvList.isNull(i));
+ } else {
+ assertFalse(tvList.isNull(i));
+ assertEquals(i, tvList.getInt(i));
+ }
}
break;
case INT64:
for (int i = 0; i < ITERATION_TIMES; ++i) {
assertEquals(i, tvList.getTime(i));
- assertEquals(i, tvList.getLong(i));
+ if (i % 7 == 0) {
+ assertTrue(tvList.isNull(i));
+ } else {
+ assertFalse(tvList.isNull(i));
+ assertEquals(i, tvList.getLong(i));
+ }
}
break;
case FLOAT:
for (int i = 0; i < ITERATION_TIMES; ++i) {
assertEquals(i, tvList.getTime(i));
- assertEquals(i, tvList.getFloat(i), 0);
+ if (i % 7 == 0) {
+ assertTrue(tvList.isNull(i));
+ } else {
+ assertFalse(tvList.isNull(i));
+ assertEquals(i, tvList.getFloat(i), 0);
+ }
}
break;
case DOUBLE:
for (int i = 0; i < ITERATION_TIMES; ++i) {
assertEquals(i, tvList.getTime(i));
- assertEquals(i, tvList.getDouble(i), 0);
+ if (i % 7 == 0) {
+ assertTrue(tvList.isNull(i));
+ } else {
+ assertFalse(tvList.isNull(i));
+ assertEquals(i, tvList.getDouble(i), 0);
+ }
}
break;
case BOOLEAN:
for (int i = 0; i < ITERATION_TIMES; ++i) {
assertEquals(i, tvList.getTime(i));
- assertEquals(i % 2 == 0, tvList.getBoolean(i));
+ if (i % 7 == 0) {
+ assertTrue(tvList.isNull(i));
+ } else {
+ assertFalse(tvList.isNull(i));
+ assertEquals(i % 2 == 0, tvList.getBoolean(i));
+ }
}
break;
case TEXT:
for (int i = 0; i < ITERATION_TIMES; ++i) {
assertEquals(i, tvList.getTime(i));
- assertEquals(Binary.valueOf(String.valueOf(i)), tvList.getBinary(i));
+ if (i % 7 == 0) {
+ assertTrue(tvList.isNull(i));
+ } else {
+ assertFalse(tvList.isNull(i));
+ assertEquals(Binary.valueOf(String.valueOf(i)), tvList.getBinary(i));
+ }
}
break;
}
@@ -196,59 +251,99 @@ public class ElasticSerializableTVListTest extends SerializableListTest {
try {
for (int i = 0; i < ITERATION_TIMES; ++i) {
- tvList.putBinary(
- i,
- Binary.valueOf(
- generateRandomString(
- byteLengthMin + random.nextInt(byteLengthMax - byteLengthMin))));
+ if (i % 7 == 0) {
+ tvList.putNull(i);
+ } else {
+ tvList.putBinary(
+ i,
+ Binary.valueOf(
+ generateRandomString(
+ byteLengthMin + random.nextInt(byteLengthMax - byteLengthMin))));
+ }
}
LayerPointReader reader = tvList.constructPointReaderUsingTrivialEvictionStrategy();
+ int index = 0;
while (reader.next()) {
- int length = reader.currentBinary().getLength();
- assertTrue(byteLengthMin <= length && length < byteLengthMax);
+ if (index % 7 == 0) {
+ assertTrue(reader.isCurrentNull());
+ } else {
+ int length = reader.currentBinary().getLength();
+ assertTrue(byteLengthMin <= length && length < byteLengthMax);
+ }
reader.readyForNext();
+ index++;
}
byteLengthMin = SerializableList.INITIAL_BYTE_ARRAY_LENGTH_FOR_MEMORY_CONTROL * 16;
byteLengthMax = SerializableList.INITIAL_BYTE_ARRAY_LENGTH_FOR_MEMORY_CONTROL * 32;
for (int i = 0; i < ITERATION_TIMES; ++i) {
- tvList.putBinary(
- i,
- Binary.valueOf(
- generateRandomString(
- byteLengthMin + random.nextInt(byteLengthMax - byteLengthMin))));
+ if (i % 7 == 0) {
+ tvList.putNull(i);
+ } else {
+ tvList.putBinary(
+ i,
+ Binary.valueOf(
+ generateRandomString(
+ byteLengthMin + random.nextInt(byteLengthMax - byteLengthMin))));
+ }
}
+ index = 0;
while (reader.next()) {
- int length = reader.currentBinary().getLength();
- assertTrue(byteLengthMin <= length && length < byteLengthMax);
+ if (index % 7 == 0) {
+ assertTrue(reader.isCurrentNull());
+ } else {
+ int length = reader.currentBinary().getLength();
+ assertTrue(byteLengthMin <= length && length < byteLengthMax);
+ }
reader.readyForNext();
+ index++;
}
byteLengthMin = SerializableList.INITIAL_BYTE_ARRAY_LENGTH_FOR_MEMORY_CONTROL * 256;
byteLengthMax = SerializableList.INITIAL_BYTE_ARRAY_LENGTH_FOR_MEMORY_CONTROL * 512;
for (int i = 0; i < ITERATION_TIMES; ++i) {
- tvList.putBinary(
- i,
- Binary.valueOf(
- generateRandomString(
- byteLengthMin + random.nextInt(byteLengthMax - byteLengthMin))));
+ if (i % 7 == 0) {
+ tvList.putNull(i);
+ } else {
+ tvList.putBinary(
+ i,
+ Binary.valueOf(
+ generateRandomString(
+ byteLengthMin + random.nextInt(byteLengthMax - byteLengthMin))));
+ }
}
+ index = 0;
while (reader.next()) {
- int length = reader.currentBinary().getLength();
- assertTrue(byteLengthMin <= length && length < byteLengthMax);
+ if (index % 7 == 0) {
+ assertTrue(reader.isCurrentNull());
+ } else {
+ int length = reader.currentBinary().getLength();
+ assertTrue(byteLengthMin <= length && length < byteLengthMax);
+ }
reader.readyForNext();
+ index++;
}
+ index = 0;
for (int i = 0; i < 2 * ITERATION_TIMES; ++i) {
- tvList.putBinary(
- i,
- Binary.valueOf(
- generateRandomString(
- byteLengthMin + random.nextInt(byteLengthMax - byteLengthMin))));
+ if (i % 7 == 0) {
+ tvList.putNull(i);
+ } else {
+ tvList.putBinary(
+ i,
+ Binary.valueOf(
+ generateRandomString(
+ byteLengthMin + random.nextInt(byteLengthMax - byteLengthMin))));
+ }
reader.next();
- int length = reader.currentBinary().getLength();
- assertTrue(byteLengthMin <= length && length < byteLengthMax);
+ if (index % 7 == 0) {
+ assertTrue(reader.isCurrentNull());
+ } else {
+ int length = reader.currentBinary().getLength();
+ assertTrue(byteLengthMin <= length && length < byteLengthMax);
+ }
reader.readyForNext();
+ index++;
}
assertEquals(ITERATION_TIMES * 5, tvList.size());
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 6744c24..51e477b 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
@@ -49,7 +49,7 @@ import static org.apache.iotdb.tsfile.utils.ReadWriteIOUtils.ClassSerializeId.NU
import static org.apache.iotdb.tsfile.utils.ReadWriteIOUtils.ClassSerializeId.STRING;
/**
- * ConverterUtils is a utility class. It provide conversion between normal datatype and byte array.
+ * ConverterUtils is a utility class. It provides conversion between normal datatype and byte array.
*/
public class ReadWriteIOUtils {
@@ -59,6 +59,7 @@ public class ReadWriteIOUtils {
public static final int LONG_LEN = 8;
public static final int DOUBLE_LEN = 8;
public static final int FLOAT_LEN = 4;
+ public static final float BIT_LEN = 0.125F;
private static final byte[] magicStringBytes;