You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ca...@apache.org on 2023/08/23 10:24:50 UTC
[iotdb] branch master updated: [IOTDB-6103] Adding count_time aggregation feature (#10756)
This is an automated email from the ASF dual-hosted git repository.
caogaofei 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 4341c152a77 [IOTDB-6103] Adding count_time aggregation feature (#10756)
4341c152a77 is described below
commit 4341c152a77265baa8e29010cfd62611b9958008
Author: Beyyes <cg...@foxmail.com>
AuthorDate: Wed Aug 23 18:24:42 2023 +0800
[IOTDB-6103] Adding count_time aggregation feature (#10756)
---
.../aggregation/IoTDBCountTimeAlignedDeviceIT.java | 304 ++++++++++++++++++
.../iotdb/db/it/aggregation/IoTDBCountTimeIT.java | 339 ++++++++++++++++++++
.../execution/aggregation/AccumulatorFactory.java | 3 +-
.../execution/aggregation/Aggregator.java | 13 +-
.../execution/aggregation/CountAccumulator.java | 1 -
...tAccumulator.java => CountTimeAccumulator.java} | 21 +-
.../SlidingWindowAggregatorFactory.java | 1 +
.../operator/process/AbstractIntoOperator.java | 4 -
.../operator/process/DeviceViewIntoOperator.java | 8 +-
.../execution/operator/process/IntoOperator.java | 9 +-
.../AbstractSeriesAggregationScanOperator.java | 10 +-
.../db/queryengine/plan/analyze/Analysis.java | 3 +-
.../queryengine/plan/analyze/AnalyzeVisitor.java | 347 ++++++++++++---------
.../plan/analyze/ExpressionAnalyzer.java | 2 +
.../plan/analyze/ExpressionTypeAnalyzer.java | 4 +-
.../queryengine/plan/analyze/ExpressionUtils.java | 3 +-
.../plan/expression/ExpressionFactory.java | 7 +
.../plan/expression/binary/BinaryExpression.java | 4 +
.../plan/expression/leaf/TimestampOperand.java | 5 +-
.../plan/expression/multi/FunctionExpression.java | 18 ++
...java => CountTimeAggregationAmountVisitor.java} | 33 +-
.../visitor/ExpressionNormalizeVisitor.java | 3 +-
.../visitor/GetMeasurementExpressionVisitor.java | 3 +-
.../visitor/LowercaseNormalizeVisitor.java | 3 +-
.../cartesian/BindSchemaForExpressionVisitor.java | 19 ++
.../cartesian/BindSchemaForPredicateVisitor.java | 18 ++
...catDeviceAndBindSchemaForExpressionVisitor.java | 23 +-
.../db/queryengine/plan/parser/ASTVisitor.java | 1 +
.../plan/planner/LogicalPlanBuilder.java | 60 +++-
.../plan/planner/LogicalPlanVisitor.java | 35 ++-
.../plan/planner/OperatorTreeGenerator.java | 71 +++--
.../plan/statement/crud/QueryStatement.java | 77 ++++-
.../schemaregion/utils/ResourceByPathUtils.java | 30 +-
.../schemaengine/schemaregion/view/ViewPaths.java | 2 +-
.../dataregion/tsfile/TsFileResource.java | 17 -
.../org/apache/iotdb/db/utils/FileLoaderUtils.java | 42 ++-
.../org/apache/iotdb/db/utils/SchemaUtils.java | 3 +
.../apache/iotdb/db/utils/TypeInferenceUtils.java | 6 +-
.../iotdb/db/utils/constant/SqlConstant.java | 3 +
.../execution/aggregation/AccumulatorTest.java | 44 +++
.../execution/operator/OperatorMemoryTest.java | 1 -
.../queryengine/plan/analyze/AnalyzeFailTest.java | 72 +++++
.../datanode1conf/iotdb-common.properties | 2 +-
.../datanode2conf/iotdb-common.properties | 3 +-
.../datanode3conf/iotdb-common.properties | 2 +-
.../udf/builtin/BuiltinAggregationFunction.java | 5 +-
.../file/metadata/AlignedTimeSeriesMetadata.java | 34 +-
.../read/reader/chunk/AlignedChunkReader.java | 4 +-
.../tsfile/read/reader/page/AlignedPageReader.java | 84 ++---
.../tsfile/write/chunk/AlignedChunkWriterImpl.java | 2 +-
.../thrift-commons/src/main/thrift/common.thrift | 3 +-
51 files changed, 1438 insertions(+), 373 deletions(-)
diff --git a/integration-test/src/test/java/org/apache/iotdb/db/it/aggregation/IoTDBCountTimeAlignedDeviceIT.java b/integration-test/src/test/java/org/apache/iotdb/db/it/aggregation/IoTDBCountTimeAlignedDeviceIT.java
new file mode 100644
index 00000000000..04b9f5746d4
--- /dev/null
+++ b/integration-test/src/test/java/org/apache/iotdb/db/it/aggregation/IoTDBCountTimeAlignedDeviceIT.java
@@ -0,0 +1,304 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.it.aggregation;
+
+import org.apache.iotdb.it.env.EnvFactory;
+import org.apache.iotdb.it.framework.IoTDBTestRunner;
+import org.apache.iotdb.itbase.category.ClusterIT;
+import org.apache.iotdb.itbase.category.LocalStandaloneIT;
+
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+
+import static org.apache.iotdb.db.it.utils.TestUtils.prepareData;
+import static org.apache.iotdb.db.it.utils.TestUtils.resultSetEqualTest;
+
+@RunWith(IoTDBTestRunner.class)
+@Category({LocalStandaloneIT.class, ClusterIT.class})
+public class IoTDBCountTimeAlignedDeviceIT {
+
+ protected static final String[] SQL_LIST =
+ new String[] {
+ // test normal query
+ "CREATE DATABASE root.aligned;",
+ "CREATE ALIGNED TIMESERIES root.aligned.db.d1(s1 INT32, s2 INT32);",
+ "CREATE ALIGNED TIMESERIES root.aligned.db.d2(s1 INT32, s2 INT32);",
+ "INSERT INTO root.aligned.db.d1(time, s1, s2) ALIGNED VALUES(1, 1, null), (2, null, 2);",
+ "INSERT INTO root.aligned.db.d2(time, s1, s2) ALIGNED VALUES(1, null, 1);",
+ // test group by time
+ "CREATE ALIGNED TIMESERIES root.aligned.downsampling.d1(s1 INT32, s2 INT32);",
+ "CREATE ALIGNED TIMESERIES root.aligned.downsampling.d2(s1 INT32, s2 INT32);",
+ "INSERT INTO root.aligned.downsampling.d1(time, s1, s2) ALIGNED VALUES(0, 0, null), (1, null, 1), "
+ + "(2, null, 2), (4,4,null), (5,5,5), (7,null,7), (8,8,8), (9,null,9);",
+ "INSERT INTO root.aligned.downsampling.d2(time, s1, s2) ALIGNED VALUES(0,null,0) (1, 1, null), (2,2,null), "
+ + "(4,null,4), (5,5,5), (7,7,null), (8,8,8);",
+
+ // test group by variation
+ "CREATE ALIGNED TIMESERIES root.aligned.variation.d1(state INT32, s1 INT32);",
+ "CREATE ALIGNED TIMESERIES root.aligned.variation.d2(state INT32, s1 INT32);",
+ "INSERT INTO root.aligned.variation.d1(time, state, s1) ALIGNED VALUES(0,0,0), (1,0,null), (2,null,2), (3,0,3), (4,0,null),(5,1,null),(6,1,6);",
+ "INSERT INTO root.aligned.variation.d2(time, state, s1) ALIGNED VALUES(0,0,null), (1,null,1), (2,1,2), (3,1,3), (4,1,null), (6,1,null);",
+ // test group by session
+ "CREATE ALIGNED TIMESERIES root.aligned.session.d1(state INT32, s1 INT32);",
+ "CREATE ALIGNED TIMESERIES root.aligned.session.d2(state INT32, s1 INT32);",
+ "INSERT INTO root.aligned.session.d1(time, state, s1) ALIGNED VALUES(0,0,0), (1,0,null), (20,0,2), (23,0,3),(40,0,null),(55,1,null),(56,1,6);",
+ "INSERT INTO root.aligned.session.d2(time, state, s1) ALIGNED VALUES(0,0,null), (1,null,1), (20,1,2), (23,1,3), (40,1,null), (56,1,null);",
+
+ // test group by condition
+ "CREATE ALIGNED TIMESERIES root.aligned.condition.d1(state INT32, s1 INT32, s2 INT32);",
+ "CREATE ALIGNED TIMESERIES root.aligned.condition.d2(state INT32, s1 INT32);",
+ "INSERT INTO root.aligned.condition.d1(time, state, s1, s2) ALIGNED VALUES(0,0,0,0), (1,1,null,1), (20,null,null,2), (23,1,3,3),(40,0,null,null),(55,1,null,null),(56,1,null,6);",
+ "INSERT INTO root.aligned.condition.d2(time, state, s1) ALIGNED VALUES(0,0,null), (1,null,1), (20,1,2), (23,1,3), (40,1,null), (56,1,null);",
+ // test having
+ "CREATE ALIGNED TIMESERIES root.aligned.having.d1(s1 INT32, s2 INT32);",
+ "CREATE ALIGNED TIMESERIES root.aligned.having.d2(s1 INT32, s2 INT32);",
+ "INSERT INTO root.aligned.having.d1(time, s1, s2) ALIGNED VALUES(0,0,null), (1,null,1), (2,null,2), (4,4,null), (5,5,5), (7,null,7), (8,8,8), (9,null,9);",
+ "INSERT INTO root.aligned.having.d2(time, s1, s2) ALIGNED VALUES(0,null,0), (1,1,null), (2,2,null), (4,null,4), (5,5,5), (7,7,null), (8,8,8), (9,9,null);",
+ };
+
+ @BeforeClass
+ public static void setUp() throws Exception {
+ // TODO set
+ EnvFactory.getEnv().getConfig().getCommonConfig().setPartitionInterval(1000);
+ EnvFactory.getEnv().initClusterEnvironment();
+ prepareData(SQL_LIST);
+ }
+
+ @AfterClass
+ public static void tearDown() throws Exception {
+ EnvFactory.getEnv().cleanClusterEnvironment();
+ }
+
+ @Test
+ public void normalQueryTest() {
+ // align by time
+ String[] expectedHeader = new String[] {"count_time(*)"};
+ String[] retArray = new String[] {"2,"};
+ resultSetEqualTest("SELECT COUNT_TIME(*) FROM root.aligned.db.**;", expectedHeader, retArray);
+
+ expectedHeader = new String[] {"count_time(*)"};
+ retArray = new String[] {"2,"};
+ resultSetEqualTest(
+ "SELECT COUNT_TIME(*) FROM root.aligned.db.d1, root.aligned.db.d2;",
+ expectedHeader,
+ retArray);
+
+ // align by device
+ expectedHeader = new String[] {"Device,count_time(*)"};
+ retArray = new String[] {"root.aligned.db.d1,2,", "root.aligned.db.d2,1,"};
+ resultSetEqualTest(
+ "select count_time(*) from root.aligned.db.** align by device;", expectedHeader, retArray);
+
+ expectedHeader = new String[] {"Device,count_time(*)"};
+ retArray = new String[] {"root.aligned.db.d1,2,", "root.aligned.db.d2,1,"};
+ resultSetEqualTest(
+ "select count_time(*) from root.aligned.db.d1,root.aligned.db.d2 align by device;",
+ expectedHeader,
+ retArray);
+ }
+
+ @Test
+ public void groupByTimeTest() {
+ // align by time
+ String[] expectedHeader = new String[] {"Time,count_time(*)"};
+ String[] retArray = new String[] {"0,2,", "2,1,", "4,2,", "6,1,", "8,2,"};
+ resultSetEqualTest(
+ "SELECT count_time(*) FROM root.aligned.downsampling.** GROUP BY([0, 10), 2ms);",
+ expectedHeader,
+ retArray);
+ resultSetEqualTest(
+ "SELECT count_time(*) FROM root.aligned.downsampling.d1,root.aligned.downsampling.d2 GROUP BY([0, 10), 2ms);",
+ expectedHeader,
+ retArray);
+ resultSetEqualTest(
+ "SELECT count_time(*) FROM root.aligned.downsampling.d2,root.aligned.downsampling.d1 GROUP BY([0, 10), 2ms);",
+ expectedHeader,
+ retArray);
+
+ expectedHeader = new String[] {"Time,count_time(*)"};
+ retArray = new String[] {"0,2,", "2,1,", "4,2,", "6,1,", "8,2,"};
+ resultSetEqualTest(
+ "SELECT count_time(*) FROM root.aligned.downsampling.d1 GROUP BY([0, 10), 2ms);",
+ expectedHeader,
+ retArray);
+
+ // align by device
+ expectedHeader = new String[] {"Time,Device,count_time(*)"};
+ retArray =
+ new String[] {
+ "0,root.aligned.downsampling.d1,2,",
+ "2,root.aligned.downsampling.d1,1,",
+ "4,root.aligned.downsampling.d1,2,",
+ "6,root.aligned.downsampling.d1,1,",
+ "8,root.aligned.downsampling.d1,2,",
+ "0,root.aligned.downsampling.d2,2,",
+ "2,root.aligned.downsampling.d2,1,",
+ "4,root.aligned.downsampling.d2,2,",
+ "6,root.aligned.downsampling.d2,1,",
+ "8,root.aligned.downsampling.d2,1,",
+ };
+ resultSetEqualTest(
+ "SELECT count_time(*) FROM root.aligned.downsampling.** GROUP BY([0, 10), 2ms) ALIGN BY DEVICE;",
+ expectedHeader,
+ retArray);
+ resultSetEqualTest(
+ "SELECT count_time(*) FROM root.aligned.downsampling.d1,root.aligned.downsampling.d2 GROUP BY([0, 10), 2ms) ALIGN BY DEVICE;",
+ expectedHeader,
+ retArray);
+ resultSetEqualTest(
+ "SELECT count_time(*) FROM root.aligned.downsampling.d2,root.aligned.downsampling.d1 GROUP BY([0, 10), 2ms) ALIGN BY DEVICE;",
+ expectedHeader,
+ retArray);
+
+ // test sort
+ expectedHeader = new String[] {"Time,Device,count_time(*)"};
+ retArray =
+ new String[] {
+ "2,root.aligned.downsampling.d1,1,",
+ "6,root.aligned.downsampling.d1,1,",
+ "2,root.aligned.downsampling.d2,1,",
+ "6,root.aligned.downsampling.d2,1,",
+ "8,root.aligned.downsampling.d2,1,",
+ "0,root.aligned.downsampling.d1,2,",
+ "4,root.aligned.downsampling.d1,2,",
+ "8,root.aligned.downsampling.d1,2,",
+ "0,root.aligned.downsampling.d2,2,",
+ "4,root.aligned.downsampling.d2,2,",
+ };
+ resultSetEqualTest(
+ "SELECT count_time(*) FROM root.aligned.downsampling.** GROUP BY([0, 10), 2ms) ORDER BY count_time(*) ALIGN BY DEVICE;",
+ expectedHeader,
+ retArray);
+ }
+
+ @Test
+ public void groupByVariationTest() {
+ // align by time
+ String[] expectedHeader = new String[] {"Time,__endTime,count_time(*)"};
+ String[] retArray = new String[] {"0,1,2,", "2,2,1,", "3,4,2,", "5,6,2,"};
+ resultSetEqualTest(
+ "select __endTime, count_time(*) from root.aligned.variation.d1 group by variation(state, 0, ignoreNull=False);",
+ expectedHeader,
+ retArray);
+
+ // align by device
+ expectedHeader = new String[] {"Time,Device,__endTime,count_time(*)"};
+ retArray =
+ new String[] {
+ "0,root.aligned.variation.d1,1,2,",
+ "2,root.aligned.variation.d1,2,1,",
+ "3,root.aligned.variation.d1,4,2,",
+ "5,root.aligned.variation.d1,6,2,",
+ "0,root.aligned.variation.d2,0,1,",
+ "1,root.aligned.variation.d2,1,1,",
+ "2,root.aligned.variation.d2,6,4,",
+ };
+ resultSetEqualTest(
+ "select __endTime, count_time(*) from root.aligned.variation.** "
+ + "group by variation(state, 0, ignoreNull=False) align by device;",
+ expectedHeader,
+ retArray);
+ resultSetEqualTest(
+ "select __endTime, count_time(*) from root.aligned.variation.d1,root.aligned.variation.d2 "
+ + "group by variation(state, 0, ignoreNull=False) align by device;",
+ expectedHeader,
+ retArray);
+ resultSetEqualTest(
+ "select __endTime, count_time(*) from root.aligned.variation.d2,root.aligned.variation.d1 "
+ + "group by variation(state, 0, ignoreNull=False) align by device;",
+ expectedHeader,
+ retArray);
+ }
+
+ @Test
+ public void groupBySessionTest() {
+ // align by time
+ String[] expectedHeader = new String[] {"Time,__endTime,count_time(*)"};
+ String[] retArray = new String[] {"0,1,2,", "20,23,2,", "40,40,1,", "55,56,2,"};
+ resultSetEqualTest(
+ "select __endTime, count_time(*) from root.aligned.session.** group by session(10ms);",
+ expectedHeader,
+ retArray);
+ resultSetEqualTest(
+ "select __endTime, count_time(*) from root.aligned.session.d1,root.aligned.session.d2 group by session(10ms);",
+ expectedHeader,
+ retArray);
+ resultSetEqualTest(
+ "select __endTime, count_time(*) from root.aligned.session.d2,root.aligned.session.d1 group by session(10ms);",
+ expectedHeader,
+ retArray);
+
+ // align by device
+ expectedHeader = new String[] {"Time,Device,__endTime,count_time(*)"};
+ retArray =
+ new String[] {
+ "0,root.aligned.session.d1,1,2,",
+ "20,root.aligned.session.d1,23,2,",
+ "40,root.aligned.session.d1,40,1,",
+ "55,root.aligned.session.d1,56,2,",
+ "0,root.aligned.session.d2,1,2,",
+ "20,root.aligned.session.d2,23,2,",
+ "40,root.aligned.session.d2,40,1,",
+ "56,root.aligned.session.d2,56,1,",
+ };
+ resultSetEqualTest(
+ "select __endTime, count_time(*) from root.aligned.session.** group by session(10ms) align by device;",
+ expectedHeader,
+ retArray);
+ resultSetEqualTest(
+ "select __endTime, count_time(*) from root.aligned.session.d1,root.aligned.session.d2 group by session(10ms) align by device;",
+ expectedHeader,
+ retArray);
+ resultSetEqualTest(
+ "select __endTime, count_time(*) from root.aligned.session.d2,root.aligned.session.d1 group by session(10ms) align by device;",
+ expectedHeader,
+ retArray);
+ }
+
+ @Test
+ public void groupByConditionTest() {
+ // align by time
+ String[] expectedHeader = new String[] {"Time,count_time(*)"};
+ String[] retArray = new String[] {"55,2,"};
+ resultSetEqualTest(
+ "select count_time(*) from root.aligned.condition.d1 group by condition(state=1, KEEP>=2, ignoreNull=false);",
+ expectedHeader,
+ retArray);
+
+ // align by device
+ expectedHeader = new String[] {"Time,Device,count_time(*)"};
+ retArray = new String[] {"55,root.aligned.condition.d1,2,", "20,root.aligned.condition.d2,4,"};
+ resultSetEqualTest(
+ "select count_time(*) from root.aligned.condition.** group by condition(state=1, KEEP>=2, ignoreNull=false) align by device;",
+ expectedHeader,
+ retArray);
+ resultSetEqualTest(
+ "select count_time(*) from root.aligned.condition.d1,root.aligned.condition.d2 group by condition(state=1, KEEP>=2, ignoreNull=false) align by device;",
+ expectedHeader,
+ retArray);
+ resultSetEqualTest(
+ "select count_time(*) from root.aligned.condition.d2,root.aligned.condition.d1 group by condition(state=1, KEEP>=2, ignoreNull=false) align by device;",
+ expectedHeader,
+ retArray);
+ }
+}
diff --git a/integration-test/src/test/java/org/apache/iotdb/db/it/aggregation/IoTDBCountTimeIT.java b/integration-test/src/test/java/org/apache/iotdb/db/it/aggregation/IoTDBCountTimeIT.java
new file mode 100644
index 00000000000..372b16356c0
--- /dev/null
+++ b/integration-test/src/test/java/org/apache/iotdb/db/it/aggregation/IoTDBCountTimeIT.java
@@ -0,0 +1,339 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.it.aggregation;
+
+import org.apache.iotdb.it.env.EnvFactory;
+import org.apache.iotdb.it.framework.IoTDBTestRunner;
+import org.apache.iotdb.itbase.category.ClusterIT;
+import org.apache.iotdb.itbase.category.LocalStandaloneIT;
+import org.apache.iotdb.rpc.TSStatusCode;
+
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+
+import static org.apache.iotdb.db.it.utils.TestUtils.assertTestFail;
+import static org.apache.iotdb.db.it.utils.TestUtils.prepareData;
+import static org.apache.iotdb.db.it.utils.TestUtils.resultSetEqualTest;
+import static org.apache.iotdb.db.queryengine.plan.expression.visitor.CountTimeAggregationAmountVisitor.COUNT_TIME_ONLY_SUPPORT_ONE_WILDCARD;
+
+@RunWith(IoTDBTestRunner.class)
+@Category({LocalStandaloneIT.class, ClusterIT.class})
+public class IoTDBCountTimeIT {
+
+ protected static final String[] SQL_LIST =
+ new String[] {
+ // test normal query
+ "CREATE DATABASE root.db;",
+ "CREATE TIMESERIES root.db.d1.s1 WITH DATATYPE=INT32, ENCODING=PLAIN;",
+ "CREATE TIMESERIES root.db.d1.s2 WITH DATATYPE=INT32, ENCODING=PLAIN;",
+ "CREATE TIMESERIES root.db.d2.s1 WITH DATATYPE=INT32, ENCODING=PLAIN;",
+ "CREATE TIMESERIES root.db.d2.s2 WITH DATATYPE=INT32, ENCODING=PLAIN;",
+ "INSERT INTO root.db.d1(time, s1) VALUES(1, 1);",
+ "INSERT INTO root.db.d1(time, s2) VALUES(2, 2);",
+ "INSERT INTO root.db.d2(time, s2) VALUES(1, 1);",
+ // test group by time
+ "CREATE DATABASE root.downsampling;",
+ "CREATE TIMESERIES root.downsampling.d1.s1 WITH DATATYPE=INT32, ENCODING=PLAIN;",
+ "CREATE TIMESERIES root.downsampling.d1.s2 WITH DATATYPE=INT32, ENCODING=PLAIN;",
+ "CREATE TIMESERIES root.downsampling.d2.s1 WITH DATATYPE=INT32, ENCODING=PLAIN;",
+ "CREATE TIMESERIES root.downsampling.d2.s2 WITH DATATYPE=INT32, ENCODING=PLAIN;",
+ "INSERT INTO root.downsampling.d1(time, s1) VALUES(0, 0), (4,4), (5,5), (8,8);",
+ "INSERT INTO root.downsampling.d1(time, s2) VALUES(1, 1), (2,2), (5,5), (7,7), (8,8), (9,9);",
+ "INSERT INTO root.downsampling.d2(time, s1) VALUES(1, 1), (2,2), (5,5), (7,7), (8,8);",
+ "INSERT INTO root.downsampling.d2(time, s2) VALUES(0, 0), (4,4), (5,5), (8,8);",
+ // test group by variation
+ "CREATE DATABASE root.variation;",
+ "CREATE TIMESERIES root.variation.d1.state WITH DATATYPE=INT32, ENCODING=PLAIN;",
+ "CREATE TIMESERIES root.variation.d1.s1 WITH DATATYPE=INT32, ENCODING=PLAIN;",
+ "CREATE TIMESERIES root.variation.d2.state WITH DATATYPE=INT32, ENCODING=PLAIN;",
+ "CREATE TIMESERIES root.variation.d2.s1 WITH DATATYPE=INT32, ENCODING=PLAIN;",
+ "INSERT INTO root.variation.d1(time, state) VALUES(0,0), (1,0), (3,0), (4,0),(5,1),(6,1);",
+ "INSERT INTO root.variation.d1(time, s1) VALUES(0,0), (2,2), (3,3), (6,6);",
+ "INSERT INTO root.variation.d2(time, state) VALUES(0,0), (2,1), (3,1), (4,1), (6,1);",
+ "INSERT INTO root.variation.d2(time, s1) VALUES(1,1), (2,2), (3,3);",
+ // test group by session
+ "CREATE DATABASE root.session;",
+ "CREATE TIMESERIES root.session.d1.state WITH DATATYPE=INT32, ENCODING=PLAIN;",
+ "CREATE TIMESERIES root.session.d1.s1 WITH DATATYPE=INT32, ENCODING=PLAIN;",
+ "CREATE TIMESERIES root.session.d2.state WITH DATATYPE=INT32, ENCODING=PLAIN;",
+ "CREATE TIMESERIES root.session.d2.s1 WITH DATATYPE=INT32, ENCODING=PLAIN;",
+ "INSERT INTO root.session.d1(time, state) VALUES(0,0), (1,0), (20,0), (23,0),(40,0),(55,1),(56,1);",
+ "INSERT INTO root.session.d1(time, s1) VALUES(0,0), (20,2), (23,3), (56,6);",
+ "INSERT INTO root.session.d2(time, state) VALUES(0,0), (20,1), (23,1), (40,1), (56,1);",
+ "INSERT INTO root.session.d2(time, s1) VALUES(1,1), (20,2), (23,3);",
+ // test group by condition
+ "CREATE DATABASE root.condition;",
+ "CREATE TIMESERIES root.condition.d1.state WITH DATATYPE=INT32, ENCODING=PLAIN;",
+ "CREATE TIMESERIES root.condition.d1.s1 WITH DATATYPE=INT32, ENCODING=PLAIN;",
+ "CREATE TIMESERIES root.condition.d1.s2 WITH DATATYPE=INT32, ENCODING=PLAIN;",
+ "CREATE TIMESERIES root.condition.d2.state WITH DATATYPE=INT32, ENCODING=PLAIN;",
+ "CREATE TIMESERIES root.condition.d2.s1 WITH DATATYPE=INT32, ENCODING=PLAIN;",
+ "INSERT INTO root.condition.d1(time, state) VALUES(0,0), (1,1), (23,1),(40,0),(55,1),(56,1);",
+ "INSERT INTO root.condition.d1(time, s1) VALUES(0,0), (23,3), (56,6);",
+ "INSERT INTO root.condition.d1(time, s2) VALUES(0,0), (1,1), (20,2), (23,3);",
+ "INSERT INTO root.condition.d2(time, state) VALUES(0,0), (20,1), (23,1), (40,1), (56,1);",
+ "INSERT INTO root.condition.d2(time, s1) VALUES(1,1), (20,2), (23,3);",
+ // test having
+ "CREATE DATABASE root.having;",
+ "CREATE TIMESERIES root.having.d1.s1 WITH DATATYPE=INT32, ENCODING=PLAIN;",
+ "CREATE TIMESERIES root.having.d1.s2 WITH DATATYPE=INT32, ENCODING=PLAIN;",
+ "CREATE TIMESERIES root.having.d2.s1 WITH DATATYPE=INT32, ENCODING=PLAIN;",
+ "CREATE TIMESERIES root.having.d2.s2 WITH DATATYPE=INT32, ENCODING=PLAIN;",
+ "INSERT INTO root.having.d1(time, s1) VALUES(0, 0), (4,4), (5,5), (8,8);",
+ "INSERT INTO root.having.d1(time, s2) VALUES(1, 1), (2,2), (5,5), (7,7), (8,8), (9,9);",
+ "INSERT INTO root.having.d2(time, s1) VALUES(1, 1), (2,2), (5,5), (7,7), (8,8), (9,9);",
+ "INSERT INTO root.having.d2(time, s2) VALUES(0, 0), (4,4), (5,5), (8,8);",
+ "FLUSH;"
+ };
+
+ @BeforeClass
+ public static void setUp() throws Exception {
+ // TODO set
+ EnvFactory.getEnv().getConfig().getCommonConfig().setPartitionInterval(1000);
+ EnvFactory.getEnv().initClusterEnvironment();
+ prepareData(SQL_LIST);
+ }
+
+ @AfterClass
+ public static void tearDown() throws Exception {
+ EnvFactory.getEnv().cleanClusterEnvironment();
+ }
+
+ @Test
+ public void normalQueryTest() {
+ // align by time
+ String[] expectedHeader = new String[] {"count_time(*)"};
+ String[] retArray = new String[] {"2,"};
+ resultSetEqualTest("SELECT COUNT_TIME(*) FROM root.db.**;", expectedHeader, retArray);
+
+ expectedHeader = new String[] {"count_time(*)"};
+ retArray = new String[] {"2,"};
+ resultSetEqualTest(
+ "SELECT COUNT_TIME(*) FROM root.db.d1, root.db.d2;", expectedHeader, retArray);
+
+ // align by device
+ expectedHeader = new String[] {"Device,count_time(*)"};
+ retArray = new String[] {"root.db.d1,2,", "root.db.d2,1,"};
+ resultSetEqualTest(
+ "select count_time(*) from root.db.** align by device;", expectedHeader, retArray);
+
+ expectedHeader = new String[] {"Device,count_time(*)"};
+ retArray = new String[] {"root.db.d1,2,", "root.db.d2,1,"};
+ resultSetEqualTest(
+ "select count_time(*) from root.db.d1,root.db.d2 align by device;",
+ expectedHeader,
+ retArray);
+ }
+
+ @Test
+ public void groupByTimeTest() {
+ // align by time
+ String[] expectedHeader = new String[] {"Time,count_time(*)"};
+ String[] retArray = new String[] {"0,2,", "2,1,", "4,2,", "6,1,", "8,2,"};
+ resultSetEqualTest(
+ "SELECT count_time(*) FROM root.downsampling.** GROUP BY([0, 10), 2ms);",
+ expectedHeader,
+ retArray);
+ resultSetEqualTest(
+ "SELECT count_time(*) FROM root.downsampling.d1,root.downsampling.d2 GROUP BY([0, 10), 2ms);",
+ expectedHeader,
+ retArray);
+ resultSetEqualTest(
+ "SELECT count_time(*) FROM root.downsampling.d2,root.downsampling.d1 GROUP BY([0, 10), 2ms);",
+ expectedHeader,
+ retArray);
+
+ expectedHeader = new String[] {"Time,count_time(*)"};
+ retArray = new String[] {"0,2,", "2,1,", "4,2,", "6,1,", "8,2,"};
+ resultSetEqualTest(
+ "SELECT count_time(*) FROM root.downsampling.d1 GROUP BY([0, 10), 2ms);",
+ expectedHeader,
+ retArray);
+
+ // align by device
+ expectedHeader = new String[] {"Time,Device,count_time(*)"};
+ retArray =
+ new String[] {
+ "0,root.downsampling.d1,2,",
+ "2,root.downsampling.d1,1,",
+ "4,root.downsampling.d1,2,",
+ "6,root.downsampling.d1,1,",
+ "8,root.downsampling.d1,2,",
+ "0,root.downsampling.d2,2,",
+ "2,root.downsampling.d2,1,",
+ "4,root.downsampling.d2,2,",
+ "6,root.downsampling.d2,1,",
+ "8,root.downsampling.d2,1,",
+ };
+ resultSetEqualTest(
+ "SELECT count_time(*) FROM root.downsampling.** GROUP BY([0, 10), 2ms) ALIGN BY DEVICE;",
+ expectedHeader,
+ retArray);
+ resultSetEqualTest(
+ "SELECT count_time(*) FROM root.downsampling.d1,root.downsampling.d2 GROUP BY([0, 10), 2ms) ALIGN BY DEVICE;",
+ expectedHeader,
+ retArray);
+ resultSetEqualTest(
+ "SELECT count_time(*) FROM root.downsampling.d2,root.downsampling.d1 GROUP BY([0, 10), 2ms) ALIGN BY DEVICE;",
+ expectedHeader,
+ retArray);
+
+ // test sort
+ expectedHeader = new String[] {"Time,Device,count_time(*)"};
+ retArray =
+ new String[] {
+ "2,root.downsampling.d1,1,",
+ "6,root.downsampling.d1,1,",
+ "2,root.downsampling.d2,1,",
+ "6,root.downsampling.d2,1,",
+ "8,root.downsampling.d2,1,",
+ "0,root.downsampling.d1,2,",
+ "4,root.downsampling.d1,2,",
+ "8,root.downsampling.d1,2,",
+ "0,root.downsampling.d2,2,",
+ "4,root.downsampling.d2,2,",
+ };
+ resultSetEqualTest(
+ "SELECT count_time(*) FROM root.downsampling.** GROUP BY([0, 10), 2ms) ORDER BY count_time(*) ALIGN BY DEVICE;",
+ expectedHeader,
+ retArray);
+ }
+
+ @Test
+ public void groupByVariationTest() {
+ // align by time
+ String[] expectedHeader = new String[] {"Time,__endTime,count_time(*)"};
+ String[] retArray = new String[] {"0,1,2,", "2,2,1,", "3,4,2,", "5,6,2,"};
+ resultSetEqualTest(
+ "select __endTime, count_time(*) from root.variation.d1 group by variation(state, 0, ignoreNull=False);",
+ expectedHeader,
+ retArray);
+
+ // align by device
+ expectedHeader = new String[] {"Time,Device,__endTime,count_time(*)"};
+ retArray =
+ new String[] {
+ "0,root.variation.d1,1,2,",
+ "2,root.variation.d1,2,1,",
+ "3,root.variation.d1,4,2,",
+ "5,root.variation.d1,6,2,",
+ "0,root.variation.d2,0,1,",
+ "1,root.variation.d2,1,1,",
+ "2,root.variation.d2,6,4,",
+ };
+ resultSetEqualTest(
+ "select __endTime, count_time(*) from root.variation.** "
+ + "group by variation(state, 0, ignoreNull=False) align by device;",
+ expectedHeader,
+ retArray);
+ resultSetEqualTest(
+ "select __endTime, count_time(*) from root.variation.d1,root.variation.d2 "
+ + "group by variation(state, 0, ignoreNull=False) align by device;",
+ expectedHeader,
+ retArray);
+ resultSetEqualTest(
+ "select __endTime, count_time(*) from root.variation.d2,root.variation.d1 "
+ + "group by variation(state, 0, ignoreNull=False) align by device;",
+ expectedHeader,
+ retArray);
+ }
+
+ @Test
+ public void groupBySessionTest() {
+ // align by time
+ String[] expectedHeader = new String[] {"Time,__endTime,count_time(*)"};
+ String[] retArray = new String[] {"0,1,2,", "20,23,2,", "40,40,1,", "55,56,2,"};
+ resultSetEqualTest(
+ "select __endTime, count_time(*) from root.session.** group by session(10ms);",
+ expectedHeader,
+ retArray);
+ resultSetEqualTest(
+ "select __endTime, count_time(*) from root.session.d1,root.session.d2 group by session(10ms);",
+ expectedHeader,
+ retArray);
+ resultSetEqualTest(
+ "select __endTime, count_time(*) from root.session.d2,root.session.d1 group by session(10ms);",
+ expectedHeader,
+ retArray);
+
+ // align by device
+ expectedHeader = new String[] {"Time,Device,__endTime,count_time(*)"};
+ retArray =
+ new String[] {
+ "0,root.session.d1,1,2,",
+ "20,root.session.d1,23,2,",
+ "40,root.session.d1,40,1,",
+ "55,root.session.d1,56,2,",
+ "0,root.session.d2,1,2,",
+ "20,root.session.d2,23,2,",
+ "40,root.session.d2,40,1,",
+ "56,root.session.d2,56,1,",
+ };
+ resultSetEqualTest(
+ "select __endTime, count_time(*) from root.session.** group by session(10ms) align by device;",
+ expectedHeader,
+ retArray);
+ resultSetEqualTest(
+ "select __endTime, count_time(*) from root.session.d1,root.session.d2 group by session(10ms) align by device;",
+ expectedHeader,
+ retArray);
+ resultSetEqualTest(
+ "select __endTime, count_time(*) from root.session.d2,root.session.d1 group by session(10ms) align by device;",
+ expectedHeader,
+ retArray);
+ }
+
+ @Test
+ public void groupByConditionTest() {
+ // align by time
+ String[] expectedHeader = new String[] {"Time,count_time(*)"};
+ String[] retArray = new String[] {"55,2,"};
+ resultSetEqualTest(
+ "select count_time(*) from root.condition.d1 group by condition(state=1, KEEP>=2, ignoreNull=false);",
+ expectedHeader,
+ retArray);
+
+ // align by device
+ expectedHeader = new String[] {"Time,Device,count_time(*)"};
+ retArray = new String[] {"55,root.condition.d1,2,", "20,root.condition.d2,4,"};
+ resultSetEqualTest(
+ "select count_time(*) from root.condition.** group by condition(state=1, KEEP>=2, ignoreNull=false) align by device;",
+ expectedHeader,
+ retArray);
+ resultSetEqualTest(
+ "select count_time(*) from root.condition.d1,root.condition.d2 group by condition(state=1, KEEP>=2, ignoreNull=false) align by device;",
+ expectedHeader,
+ retArray);
+ resultSetEqualTest(
+ "select count_time(*) from root.condition.d2,root.condition.d1 group by condition(state=1, KEEP>=2, ignoreNull=false) align by device;",
+ expectedHeader,
+ retArray);
+ }
+
+ @Test
+ public void testUnSupportedSql() {
+ assertTestFail(
+ "SELECT COUNT_TIME(s1) FROM root.db.**;",
+ TSStatusCode.SEMANTIC_ERROR.getStatusCode() + ": " + COUNT_TIME_ONLY_SUPPORT_ONE_WILDCARD);
+ }
+}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/aggregation/AccumulatorFactory.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/aggregation/AccumulatorFactory.java
index 6527bd6af43..9fdf2fbe31b 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/aggregation/AccumulatorFactory.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/aggregation/AccumulatorFactory.java
@@ -31,7 +31,6 @@ import java.util.Map;
public class AccumulatorFactory {
- // TODO: Are we going to create different seriesScanOperator based on order by sequence?
public static Accumulator createAccumulator(
TAggregationType aggregationType,
TSDataType tsDataType,
@@ -71,6 +70,8 @@ public class AccumulatorFactory {
return new TimeDurationAccumulator();
case MODE:
return crateModeAccumulator(tsDataType);
+ case COUNT_TIME:
+ return new CountTimeAccumulator();
default:
throw new IllegalArgumentException("Invalid Aggregation function: " + aggregationType);
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/aggregation/Aggregator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/aggregation/Aggregator.java
index d34e67cf7e8..f6f46c5dd0b 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/aggregation/Aggregator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/aggregation/Aggregator.java
@@ -42,7 +42,7 @@ public class Aggregator {
// In some intermediate result input, inputLocation[] should include two columns
protected List<InputLocation[]> inputLocationList;
protected final AggregationStep step;
- protected final QueryExecutionMetricSet QUERY_EXECUTION_METRICS =
+ protected static final QueryExecutionMetricSet QUERY_EXECUTION_METRICS =
QueryExecutionMetricSet.getInstance();
// Used for SeriesAggregateScanOperator
@@ -53,7 +53,7 @@ public class Aggregator {
Collections.singletonList(new InputLocation[] {new InputLocation(0, 0)});
}
- // Used for aggregateOperator
+ // Used for AggregateOperator, AlignedSeriesAggregateScanOperator
public Aggregator(
Accumulator accumulator, AggregationStep step, List<InputLocation[]> inputLocationList) {
this.accumulator = accumulator;
@@ -74,7 +74,9 @@ public class Aggregator {
"RawDataAggregateOperator can only process one tsBlock input.");
Column[] timeAndValueColumn = new Column[2];
timeAndValueColumn[0] = tsBlock.getTimeColumn();
- timeAndValueColumn[1] = tsBlock.getColumn(inputLocations[0].getValueColumnIndex());
+ int index = inputLocations[0].getValueColumnIndex();
+ // for count_time, time column is also its value column
+ timeAndValueColumn[1] = index == -1 ? timeAndValueColumn[0] : tsBlock.getColumn(index);
accumulator.addInput(timeAndValueColumn, bitMap, lastIndex);
}
} finally {
@@ -120,12 +122,13 @@ public class Aggregator {
}
/** Used for SeriesAggregateScanOperator. */
- public void processStatistics(Statistics[] statistics) {
+ public void processStatistics(Statistics timeStatistics, Statistics[] valueStatistics) {
long startTime = System.nanoTime();
try {
for (InputLocation[] inputLocations : inputLocationList) {
int valueIndex = inputLocations[0].getValueColumnIndex();
- accumulator.addStatistics(statistics[valueIndex]);
+ // valueIndex == -1 means it is count_time, we need to use timeStatistics
+ accumulator.addStatistics(valueIndex == -1 ? timeStatistics : valueStatistics[valueIndex]);
}
} finally {
QUERY_EXECUTION_METRICS.recordExecutionCost(
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/aggregation/CountAccumulator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/aggregation/CountAccumulator.java
index 1de4da9f198..870b8fbb47b 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/aggregation/CountAccumulator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/aggregation/CountAccumulator.java
@@ -36,7 +36,6 @@ public class CountAccumulator implements Accumulator {
}
// Column should be like: | Time | Value |
-
@Override
public void addInput(Column[] column, BitMap bitMap, int lastIndex) {
int curPositionCount = column[0].getPositionCount();
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/aggregation/CountAccumulator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/aggregation/CountTimeAccumulator.java
similarity index 84%
copy from iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/aggregation/CountAccumulator.java
copy to iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/aggregation/CountTimeAccumulator.java
index 1de4da9f198..6cffec5e5d0 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/aggregation/CountAccumulator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/aggregation/CountTimeAccumulator.java
@@ -7,7 +7,7 @@
* "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
+ * 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
@@ -27,30 +27,23 @@ import org.apache.iotdb.tsfile.utils.BitMap;
import static com.google.common.base.Preconditions.checkArgument;
-public class CountAccumulator implements Accumulator {
+public class CountTimeAccumulator implements Accumulator {
private long countValue = 0;
- public CountAccumulator() {
+ public CountTimeAccumulator() {
// do nothing
}
- // Column should be like: | Time | Value |
-
+ // Column should be like: | Time | Time |
@Override
public void addInput(Column[] column, BitMap bitMap, int lastIndex) {
- int curPositionCount = column[0].getPositionCount();
- if (!column[1].mayHaveNull()
- && lastIndex == curPositionCount - 1
- && ((bitMap == null) || bitMap.isAllMarked())) {
- countValue += curPositionCount;
+ if ((bitMap == null) || bitMap.isAllMarked()) {
+ countValue += (lastIndex + 1);
} else {
for (int i = 0; i <= lastIndex; i++) {
- if (bitMap != null && !bitMap.isMarked(i)) {
- continue;
- }
- if (!column[1].isNull(i)) {
+ if (bitMap.isMarked(i)) {
countValue++;
}
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/aggregation/slidingwindow/SlidingWindowAggregatorFactory.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/aggregation/slidingwindow/SlidingWindowAggregatorFactory.java
index 14e17afbb5d..53fba75c504 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/aggregation/slidingwindow/SlidingWindowAggregatorFactory.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/aggregation/slidingwindow/SlidingWindowAggregatorFactory.java
@@ -132,6 +132,7 @@ public class SlidingWindowAggregatorFactory {
case SUM:
case AVG:
case COUNT:
+ case COUNT_TIME:
return new SmoothQueueSlidingWindowAggregator(accumulator, inputLocationList, step);
case MAX_VALUE:
return new MonotonicQueueSlidingWindowAggregator(
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/AbstractIntoOperator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/AbstractIntoOperator.java
index 3401d621744..dd0f889514c 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/AbstractIntoOperator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/AbstractIntoOperator.java
@@ -66,8 +66,6 @@ public abstract class AbstractIntoOperator implements ProcessOperator {
protected List<InsertTabletStatementGenerator> insertTabletStatementGenerators;
- protected final Map<String, InputLocation> sourceColumnToInputLocationMap;
-
private DataNodeInternalClient client;
private final ExecutorService writeOperationExecutor;
@@ -85,7 +83,6 @@ public abstract class AbstractIntoOperator implements ProcessOperator {
OperatorContext operatorContext,
Operator child,
List<TSDataType> inputColumnTypes,
- Map<String, InputLocation> sourceColumnToInputLocationMap,
ExecutorService intoOperationExecutor,
long statementSizePerLine) {
this.operatorContext = operatorContext;
@@ -93,7 +90,6 @@ public abstract class AbstractIntoOperator implements ProcessOperator {
this.typeConvertors =
inputColumnTypes.stream().map(TypeFactory::getType).collect(Collectors.toList());
- this.sourceColumnToInputLocationMap = sourceColumnToInputLocationMap;
this.writeOperationExecutor = intoOperationExecutor;
initMemoryEstimates(statementSizePerLine);
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/DeviceViewIntoOperator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/DeviceViewIntoOperator.java
index 44bf02ad940..9d9d7e3b1d3 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/DeviceViewIntoOperator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/DeviceViewIntoOperator.java
@@ -67,13 +67,7 @@ public class DeviceViewIntoOperator extends AbstractIntoOperator {
Map<String, InputLocation> sourceColumnToInputLocationMap,
ExecutorService intoOperationExecutor,
long statementSizePerLine) {
- super(
- operatorContext,
- child,
- inputColumnTypes,
- sourceColumnToInputLocationMap,
- intoOperationExecutor,
- statementSizePerLine);
+ super(operatorContext, child, inputColumnTypes, intoOperationExecutor, statementSizePerLine);
this.deviceToTargetPathSourceInputLocationMap = deviceToTargetPathSourceInputLocationMap;
this.deviceToTargetPathDataTypeMap = deviceToTargetPathDataTypeMap;
this.targetDeviceToAlignedMap = targetDeviceToAlignedMap;
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/IntoOperator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/IntoOperator.java
index 8a6fac95b00..d389062862f 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/IntoOperator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/IntoOperator.java
@@ -51,16 +51,9 @@ public class IntoOperator extends AbstractIntoOperator {
Map<PartialPath, Map<String, TSDataType>> targetPathToDataTypeMap,
Map<String, Boolean> targetDeviceToAlignedMap,
List<Pair<String, PartialPath>> sourceTargetPathPairList,
- Map<String, InputLocation> sourceColumnToInputLocationMap,
ExecutorService intoOperationExecutor,
long statementSizePerLine) {
- super(
- operatorContext,
- child,
- inputColumnTypes,
- sourceColumnToInputLocationMap,
- intoOperationExecutor,
- statementSizePerLine);
+ super(operatorContext, child, inputColumnTypes, intoOperationExecutor, statementSizePerLine);
this.sourceTargetPathPairList = sourceTargetPathPairList;
insertTabletStatementGenerators =
constructInsertTabletStatementGenerators(
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/AbstractSeriesAggregationScanOperator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/AbstractSeriesAggregationScanOperator.java
index 89b35bd8986..482c959c911 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/AbstractSeriesAggregationScanOperator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/AbstractSeriesAggregationScanOperator.java
@@ -203,12 +203,12 @@ public abstract class AbstractSeriesAggregationScanOperator extends AbstractData
return calcResult.getLeft();
}
- protected void calcFromStatistics(Statistics[] statistics) {
+ protected void calcFromStatistics(Statistics timeStatistics, Statistics[] valueStatistics) {
for (Aggregator aggregator : aggregators) {
if (aggregator.hasFinalResult()) {
continue;
}
- aggregator.processStatistics(statistics);
+ aggregator.processStatistics(timeStatistics, valueStatistics);
}
}
@@ -232,7 +232,7 @@ public abstract class AbstractSeriesAggregationScanOperator extends AbstractData
for (int i = 0; i < subSensorSize; i++) {
statisticsList[i] = seriesScanUtil.currentFileStatistics(i);
}
- calcFromStatistics(statisticsList);
+ calcFromStatistics(fileTimeStatistics, statisticsList);
seriesScanUtil.skipCurrentFile();
if (isAllAggregatorsHasFinalResult(aggregators) && !isGroupByQuery) {
return true;
@@ -272,7 +272,7 @@ public abstract class AbstractSeriesAggregationScanOperator extends AbstractData
for (int i = 0; i < subSensorSize; i++) {
statisticsList[i] = seriesScanUtil.currentChunkStatistics(i);
}
- calcFromStatistics(statisticsList);
+ calcFromStatistics(chunkTimeStatistics, statisticsList);
seriesScanUtil.skipCurrentChunk();
if (isAllAggregatorsHasFinalResult(aggregators) && !isGroupByQuery) {
return true;
@@ -311,7 +311,7 @@ public abstract class AbstractSeriesAggregationScanOperator extends AbstractData
for (int i = 0; i < subSensorSize; i++) {
statisticsList[i] = seriesScanUtil.currentPageStatistics(i);
}
- calcFromStatistics(statisticsList);
+ calcFromStatistics(pageTimeStatistics, statisticsList);
seriesScanUtil.skipCurrentPage();
if (isAllAggregatorsHasFinalResult(aggregators) && !isGroupByQuery) {
return true;
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/Analysis.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/Analysis.java
index 0f9c4d223c6..c49ee5a0ee6 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/Analysis.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/Analysis.java
@@ -50,6 +50,7 @@ import org.apache.iotdb.tsfile.utils.Pair;
import java.util.ArrayList;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
@@ -93,7 +94,7 @@ public class Analysis {
private Set<Expression> sourceExpressions;
// input expressions of aggregations to be calculated
- private Set<Expression> sourceTransformExpressions;
+ private Set<Expression> sourceTransformExpressions = new HashSet<>();
private Expression whereExpression;
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/AnalyzeVisitor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/AnalyzeVisitor.java
index 92d2e85c1b9..7701e6b8dda 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/AnalyzeVisitor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/AnalyzeVisitor.java
@@ -180,12 +180,20 @@ import static org.apache.iotdb.db.queryengine.common.header.ColumnHeaderConstant
import static org.apache.iotdb.db.queryengine.common.header.ColumnHeaderConstant.ENDTIME;
import static org.apache.iotdb.db.queryengine.metric.QueryPlanCostMetricSet.PARTITION_FETCHER;
import static org.apache.iotdb.db.queryengine.metric.QueryPlanCostMetricSet.SCHEMA_FETCHER;
+import static org.apache.iotdb.db.queryengine.plan.analyze.ExpressionAnalyzer.bindSchemaForExpression;
+import static org.apache.iotdb.db.queryengine.plan.analyze.ExpressionAnalyzer.concatDeviceAndBindSchemaForExpression;
+import static org.apache.iotdb.db.queryengine.plan.analyze.ExpressionAnalyzer.getMeasurementExpression;
+import static org.apache.iotdb.db.queryengine.plan.analyze.ExpressionAnalyzer.normalizeExpression;
+import static org.apache.iotdb.db.queryengine.plan.analyze.ExpressionAnalyzer.searchAggregationExpressions;
+import static org.apache.iotdb.db.queryengine.plan.analyze.ExpressionAnalyzer.searchSourceExpressions;
+import static org.apache.iotdb.db.queryengine.plan.analyze.ExpressionAnalyzer.toLowerCaseExpression;
import static org.apache.iotdb.db.queryengine.plan.analyze.SelectIntoUtils.constructTargetDevice;
import static org.apache.iotdb.db.queryengine.plan.analyze.SelectIntoUtils.constructTargetMeasurement;
import static org.apache.iotdb.db.queryengine.plan.analyze.SelectIntoUtils.constructTargetPath;
import static org.apache.iotdb.db.queryengine.plan.optimization.LimitOffsetPushDown.canPushDownLimitOffsetInGroupByTimeForDevice;
import static org.apache.iotdb.db.queryengine.plan.optimization.LimitOffsetPushDown.pushDownLimitOffsetInGroupByTimeForDevice;
import static org.apache.iotdb.db.schemaengine.schemaregion.view.visitor.GetSourcePathsVisitor.getSourcePaths;
+import static org.apache.iotdb.db.utils.constant.SqlConstant.COUNT_TIME_HEADER;
/** This visitor is used to analyze each type of Statement and returns the {@link Analysis}. */
public class AnalyzeVisitor extends StatementVisitor<Analysis, MPPQueryContext> {
@@ -194,10 +202,10 @@ public class AnalyzeVisitor extends StatementVisitor<Analysis, MPPQueryContext>
private static final IoTDBConfig CONFIG = IoTDBDescriptor.getInstance().getConfig();
- private static final Expression deviceExpression =
+ private static final Expression DEVICE_EXPRESSION =
TimeSeriesOperand.constructColumnHeaderExpression(DEVICE, TSDataType.TEXT);
- private static final Expression endTimeExpression =
+ private static final Expression END_TIME_EXPRESSION =
TimeSeriesOperand.constructColumnHeaderExpression(ENDTIME, TSDataType.INT64);
private final List<String> lastQueryColumnNames =
@@ -289,6 +297,7 @@ public class AnalyzeVisitor extends StatementVisitor<Analysis, MPPQueryContext>
}
analyzeGroupBy(analysis, queryStatement, schemaTree);
+
analyzeHaving(analysis, queryStatement, schemaTree);
analyzeOrderBy(analysis, queryStatement, schemaTree);
@@ -297,17 +306,18 @@ public class AnalyzeVisitor extends StatementVisitor<Analysis, MPPQueryContext>
Set<Expression> selectExpressions = new LinkedHashSet<>();
if (queryStatement.isOutputEndTime()) {
- selectExpressions.add(endTimeExpression);
+ selectExpressions.add(END_TIME_EXPRESSION);
}
for (Pair<Expression, String> outputExpressionAndAlias : outputExpressions) {
- selectExpressions.add(outputExpressionAndAlias.left);
+ Expression outputExpression = outputExpressionAndAlias.left;
+ selectExpressions.add(outputExpression);
}
analysis.setSelectExpressions(selectExpressions);
analyzeAggregation(analysis, queryStatement);
analyzeWhere(analysis, queryStatement, schemaTree);
- analyzeSourceTransform(analysis, queryStatement);
+ analyzeSourceTransform(analysis, outputExpressions, queryStatement);
analyzeSource(analysis, queryStatement);
@@ -430,8 +440,7 @@ public class AnalyzeVisitor extends StatementVisitor<Analysis, MPPQueryContext>
sourceExpressions = new LinkedHashSet<>();
for (Expression selectExpression : selectExpressions) {
- for (Expression sourceExpression :
- ExpressionAnalyzer.bindSchemaForExpression(selectExpression, schemaTree)) {
+ for (Expression sourceExpression : bindSchemaForExpression(selectExpression, schemaTree)) {
if (!(sourceExpression instanceof TimeSeriesOperand)) {
throw new SemanticException(
"Views with functions and expressions cannot be used in LAST query");
@@ -485,6 +494,7 @@ public class AnalyzeVisitor extends StatementVisitor<Analysis, MPPQueryContext>
}
}
+ /** process select component for align by time. */
private Map<Integer, List<Pair<Expression, String>>> analyzeSelect(
Analysis analysis, QueryStatement queryStatement, ISchemaTree schemaTree) {
Map<Integer, List<Pair<Expression, String>>> outputExpressionMap = new HashMap<>();
@@ -502,35 +512,38 @@ public class AnalyzeVisitor extends StatementVisitor<Analysis, MPPQueryContext>
for (ResultColumn resultColumn : queryStatement.getSelectComponent().getResultColumns()) {
List<Pair<Expression, String>> outputExpressions = new ArrayList<>();
-
List<Expression> resultExpressions =
- ExpressionAnalyzer.bindSchemaForExpression(resultColumn.getExpression(), schemaTree);
- for (Expression expression : resultExpressions) {
+ bindSchemaForExpression(resultColumn.getExpression(), schemaTree);
+
+ for (Expression resultExpression : resultExpressions) {
if (paginationController.hasCurOffset()) {
paginationController.consumeOffset();
} else if (paginationController.hasCurLimit()) {
if (isGroupByLevel) {
- analyzeExpressionType(analysis, expression);
- outputExpressions.add(new Pair<>(expression, resultColumn.getAlias()));
+ analyzeExpressionType(analysis, resultExpression);
+ outputExpressions.add(new Pair<>(resultExpression, resultColumn.getAlias()));
queryStatement
.getGroupByLevelComponent()
.updateIsCountStar(resultColumn.getExpression());
} else {
- Expression normalizedExpression = ExpressionAnalyzer.normalizeExpression(expression);
+ Expression normalizedExpression = normalizeExpression(resultExpression);
analyzeExpressionType(analysis, normalizedExpression);
-
checkAliasUniqueness(resultColumn.getAlias(), aliasSet);
-
outputExpressions.add(
new Pair<>(
normalizedExpression,
- analyzeAlias(resultColumn.getAlias(), expression, normalizedExpression)));
+ analyzeAlias(
+ resultColumn.getAlias(),
+ resultExpression,
+ normalizedExpression,
+ queryStatement)));
}
paginationController.consumeLimit();
} else {
break;
}
}
+
outputExpressionMap.put(columnIndex++, outputExpressions);
}
return outputExpressionMap;
@@ -554,6 +567,7 @@ public class AnalyzeVisitor extends StatementVisitor<Analysis, MPPQueryContext>
: deviceSet.stream().sorted(Comparator.reverseOrder()).collect(Collectors.toList());
}
+ /** process select component for align by device. */
private List<Pair<Expression, String>> analyzeSelect(
Analysis analysis,
QueryStatement queryStatement,
@@ -561,7 +575,6 @@ public class AnalyzeVisitor extends StatementVisitor<Analysis, MPPQueryContext>
List<PartialPath> deviceSet) {
List<Pair<Expression, String>> outputExpressions = new ArrayList<>();
Map<String, Set<Expression>> deviceToSelectExpressions = new HashMap<>();
-
ColumnPaginationController paginationController =
new ColumnPaginationController(
queryStatement.getSeriesLimit(), queryStatement.getSeriesOffset(), false);
@@ -575,22 +588,21 @@ public class AnalyzeVisitor extends StatementVisitor<Analysis, MPPQueryContext>
new LinkedHashMap<>();
for (PartialPath device : deviceSet) {
List<Expression> selectExpressionsOfOneDevice =
- ExpressionAnalyzer.concatDeviceAndBindSchemaForExpression(
- selectExpression, device, schemaTree);
+ concatDeviceAndBindSchemaForExpression(selectExpression, device, schemaTree);
if (selectExpressionsOfOneDevice.isEmpty()) {
continue;
}
+
updateMeasurementToDeviceSelectExpressions(
analysis, measurementToDeviceSelectExpressions, device, selectExpressionsOfOneDevice);
}
checkAliasUniqueness(resultColumn.getAlias(), measurementToDeviceSelectExpressions);
- for (Map.Entry<Expression, Map<String, Expression>> measurementDeviceSelectExpressionsEntry :
+ for (Map.Entry<Expression, Map<String, Expression>> entry :
measurementToDeviceSelectExpressions.entrySet()) {
- Expression measurementExpression = measurementDeviceSelectExpressionsEntry.getKey();
- Map<String, Expression> deviceToSelectExpressionsOfOneMeasurement =
- measurementDeviceSelectExpressionsEntry.getValue();
+ Expression measurementExpression = entry.getKey();
+ Map<String, Expression> deviceToSelectExpressionsOfOneMeasurement = entry.getValue();
if (paginationController.hasCurOffset()) {
paginationController.consumeOffset();
@@ -604,16 +616,17 @@ public class AnalyzeVisitor extends StatementVisitor<Analysis, MPPQueryContext>
analysis, new ArrayList<>(deviceToSelectExpressionsOfOneMeasurement.values()));
// add outputExpressions
- Expression normalizedMeasurementExpression =
- ExpressionAnalyzer.toLowerCaseExpression(measurementExpression);
- analyzeExpressionType(analysis, normalizedMeasurementExpression);
+ Expression lowerCaseMeasurementExpression = toLowerCaseExpression(measurementExpression);
+ analyzeExpressionType(analysis, lowerCaseMeasurementExpression);
+
outputExpressions.add(
new Pair<>(
- normalizedMeasurementExpression,
+ lowerCaseMeasurementExpression,
analyzeAlias(
resultColumn.getAlias(),
measurementExpression,
- normalizedMeasurementExpression)));
+ lowerCaseMeasurementExpression,
+ queryStatement)));
// add deviceToSelectExpressions
updateDeviceToSelectExpressions(
@@ -642,20 +655,16 @@ public class AnalyzeVisitor extends StatementVisitor<Analysis, MPPQueryContext>
devicePath -> analysis.getDeviceToWhereExpression().remove(devicePath.getFullPath()));
}
- analysis.setDeviceToSelectExpressions(deviceToSelectExpressions);
-
- // set selectExpressions
Set<Expression> selectExpressions = new LinkedHashSet<>();
- selectExpressions.add(deviceExpression);
+ selectExpressions.add(DEVICE_EXPRESSION);
if (queryStatement.isOutputEndTime()) {
- selectExpressions.add(endTimeExpression);
+ selectExpressions.add(END_TIME_EXPRESSION);
}
- selectExpressions.addAll(
- outputExpressions.stream()
- .map(Pair::getLeft)
- .collect(Collectors.toCollection(LinkedHashSet::new)));
+ outputExpressions.forEach(pair -> selectExpressions.add(pair.getLeft()));
analysis.setSelectExpressions(selectExpressions);
+ analysis.setDeviceToSelectExpressions(deviceToSelectExpressions);
+
return outputExpressions;
}
@@ -677,29 +686,38 @@ public class AnalyzeVisitor extends StatementVisitor<Analysis, MPPQueryContext>
Analysis analysis,
Map<String, Set<Expression>> deviceToSelectExpressions,
Map<String, Expression> deviceToSelectExpressionsOfOneMeasurement) {
- for (Map.Entry<String, Expression> deviceNameSelectExpressionEntry :
+
+ for (Map.Entry<String, Expression> entry :
deviceToSelectExpressionsOfOneMeasurement.entrySet()) {
- String deviceName = deviceNameSelectExpressionEntry.getKey();
- Expression expression = deviceNameSelectExpressionEntry.getValue();
+ String deviceName = entry.getKey();
+ Expression expression = entry.getValue();
- Expression normalizedExpression = ExpressionAnalyzer.toLowerCaseExpression(expression);
- analyzeExpressionType(analysis, normalizedExpression);
+ Expression lowerCaseExpression = toLowerCaseExpression(expression);
+ analyzeExpressionType(analysis, lowerCaseExpression);
deviceToSelectExpressions
.computeIfAbsent(deviceName, key -> new LinkedHashSet<>())
- .add(normalizedExpression);
+ .add(lowerCaseExpression);
}
}
private String analyzeAlias(
- String resultColumnAlias, Expression rawExpression, Expression normalizedExpression) {
+ String resultColumnAlias,
+ Expression rawExpression,
+ Expression normalizedExpression,
+ QueryStatement queryStatement) {
if (resultColumnAlias != null) {
// use alias as output symbol
return resultColumnAlias;
}
+ if (queryStatement.isCountTimeAggregation()) {
+ return COUNT_TIME_HEADER;
+ }
+
if (!Objects.equals(normalizedExpression, rawExpression)) {
return rawExpression.getOutputSymbol();
}
+
return null;
}
@@ -719,7 +737,8 @@ public class AnalyzeVisitor extends StatementVisitor<Analysis, MPPQueryContext>
Expression havingExpression =
ExpressionUtils.constructQueryFilter(
conJunctions.stream().distinct().collect(Collectors.toList()));
- havingExpression = ExpressionAnalyzer.normalizeExpression(havingExpression);
+ havingExpression = normalizeExpression(havingExpression);
+
TSDataType outputType = analyzeExpressionType(analysis, havingExpression);
if (outputType != TSDataType.BOOLEAN) {
throw new SemanticException(
@@ -727,6 +746,7 @@ public class AnalyzeVisitor extends StatementVisitor<Analysis, MPPQueryContext>
"The output type of the expression in HAVING clause should be BOOLEAN, actual data type: %s.",
outputType));
}
+
analysis.setHavingExpression(havingExpression);
}
@@ -750,21 +770,18 @@ public class AnalyzeVisitor extends StatementVisitor<Analysis, MPPQueryContext>
for (PartialPath device : deviceSet) {
List<Expression> expressionsInHaving =
- ExpressionAnalyzer.concatDeviceAndBindSchemaForExpression(
- havingExpression, device, schemaTree);
+ concatDeviceAndBindSchemaForExpression(havingExpression, device, schemaTree);
conJunctions.addAll(
expressionsInHaving.stream()
- .map(expression -> ExpressionAnalyzer.getMeasurementExpression(expression, analysis))
+ .map(expression -> getMeasurementExpression(expression, analysis))
.collect(Collectors.toList()));
for (Expression expression : expressionsInHaving) {
Set<Expression> aggregationExpressions = new LinkedHashSet<>();
Set<Expression> normalizedAggregationExpressions = new LinkedHashSet<>();
- for (Expression aggregationExpression :
- ExpressionAnalyzer.searchAggregationExpressions(expression)) {
- Expression normalizedAggregationExpression =
- ExpressionAnalyzer.normalizeExpression(aggregationExpression);
+ for (Expression aggregationExpression : searchAggregationExpressions(expression)) {
+ Expression normalizedAggregationExpression = normalizeExpression(aggregationExpression);
analyzeExpressionType(analysis, aggregationExpression);
analyzeExpressionType(analysis, normalizedAggregationExpression);
@@ -841,8 +858,7 @@ public class AnalyzeVisitor extends StatementVisitor<Analysis, MPPQueryContext>
if (paginationController.hasCurOffset()) {
paginationController.consumeOffset();
} else if (paginationController.hasCurLimit()) {
- Expression normalizedGroupedExpression =
- ExpressionAnalyzer.normalizeExpression(groupedExpression);
+ Expression normalizedGroupedExpression = normalizeExpression(groupedExpression);
analyzeExpressionType(analysis, normalizedGroupedExpression);
outputExpressions.add(
new Pair<>(
@@ -850,7 +866,8 @@ public class AnalyzeVisitor extends StatementVisitor<Analysis, MPPQueryContext>
analyzeAlias(
groupByLevelController.getAlias(groupedExpression.getExpressionString()),
groupedExpression,
- normalizedGroupedExpression)));
+ normalizedGroupedExpression,
+ queryStatement)));
updateGroupByLevelExpressions(
analysis,
groupedExpression,
@@ -890,14 +907,13 @@ public class AnalyzeVisitor extends StatementVisitor<Analysis, MPPQueryContext>
Expression expression,
Map<Expression, Set<Expression>> groupByLevelExpressions,
Map<Expression, Set<Expression>> groupedExpressionToRawExpressionsMap) {
- for (Expression groupedAggregationExpression :
- ExpressionAnalyzer.searchAggregationExpressions(expression)) {
+ for (Expression groupedAggregationExpression : searchAggregationExpressions(expression)) {
Set<Expression> groupedExpressionSet =
groupedExpressionToRawExpressionsMap.get(groupedAggregationExpression).stream()
.map(ExpressionAnalyzer::normalizeExpression)
.collect(Collectors.toSet());
Expression groupedAggregationExpressionWithoutAlias =
- ExpressionAnalyzer.normalizeExpression(groupedAggregationExpression);
+ normalizeExpression(groupedAggregationExpression);
analyzeExpressionType(analysis, groupedAggregationExpressionWithoutAlias);
groupedExpressionSet.forEach(
@@ -934,7 +950,7 @@ public class AnalyzeVisitor extends StatementVisitor<Analysis, MPPQueryContext>
for (Pair<Expression, String> outputExpressionAndAlias : outputExpressions) {
FunctionExpression rawExpression = (FunctionExpression) outputExpressionAndAlias.getLeft();
FunctionExpression measurementExpression =
- (FunctionExpression) ExpressionAnalyzer.getMeasurementExpression(rawExpression, analysis);
+ (FunctionExpression) getMeasurementExpression(rawExpression, analysis);
outputExpressionToRawExpressionsMap
.computeIfAbsent(measurementExpression, v -> new HashSet<>())
.add(rawExpression);
@@ -978,6 +994,7 @@ public class AnalyzeVisitor extends StatementVisitor<Analysis, MPPQueryContext>
updateDeviceToAggregationAndOutputExpressions(
analysis, analysis.getDeviceToSelectExpressions());
+
if (queryStatement.hasOrderByExpression()) {
updateDeviceToAggregationAndOutputExpressions(
analysis, analysis.getDeviceToOrderByExpressions());
@@ -997,25 +1014,20 @@ public class AnalyzeVisitor extends StatementVisitor<Analysis, MPPQueryContext>
String deviceName = deviceExpressionsEntry.getKey();
Set<Expression> expressionSet = deviceExpressionsEntry.getValue();
- Set<Expression> aggregationExpressions = new LinkedHashSet<>();
- Set<Expression> normalizedAggregationExpressions = new LinkedHashSet<>();
for (Expression expression : expressionSet) {
- for (Expression aggregationExpression :
- ExpressionAnalyzer.searchAggregationExpressions(expression)) {
- Expression normalizedAggregationExpression =
- ExpressionAnalyzer.normalizeExpression(aggregationExpression);
+ for (Expression aggregationExpression : searchAggregationExpressions(expression)) {
+ Expression normalizedAggregationExpression = normalizeExpression(aggregationExpression);
+
analyzeExpressionType(analysis, normalizedAggregationExpression);
- aggregationExpressions.add(aggregationExpression);
- normalizedAggregationExpressions.add(normalizedAggregationExpression);
+ deviceToOutputExpressions
+ .computeIfAbsent(deviceName, key -> new LinkedHashSet<>())
+ .add(aggregationExpression);
+ deviceToAggregationExpressions
+ .computeIfAbsent(deviceName, key -> new LinkedHashSet<>())
+ .add(normalizedAggregationExpression);
}
}
- deviceToOutputExpressions
- .computeIfAbsent(deviceName, key -> new LinkedHashSet<>())
- .addAll(aggregationExpressions);
- deviceToAggregationExpressions
- .computeIfAbsent(deviceName, key -> new LinkedHashSet<>())
- .addAll(normalizedAggregationExpressions);
}
}
@@ -1030,50 +1042,61 @@ public class AnalyzeVisitor extends StatementVisitor<Analysis, MPPQueryContext>
.flatMap(Set::stream)
.collect(Collectors.toSet());
analysis.setAggregationExpressions(aggregationExpressions);
- } else {
- Set<Expression> aggregationExpressions = new HashSet<>();
- for (Expression expression : analysis.getSelectExpressions()) {
- aggregationExpressions.addAll(ExpressionAnalyzer.searchAggregationExpressions(expression));
- }
- if (queryStatement.hasHaving()) {
- aggregationExpressions.addAll(
- ExpressionAnalyzer.searchAggregationExpressions(analysis.getHavingExpression()));
- }
- if (queryStatement.hasOrderByExpression()) {
- for (Expression expression : analysis.getOrderByExpressions()) {
- aggregationExpressions.addAll(
- ExpressionAnalyzer.searchAggregationExpressions(expression));
- }
+ return;
+ }
+
+ Set<Expression> aggregationExpressions = new HashSet<>();
+ for (Expression expression : analysis.getSelectExpressions()) {
+ aggregationExpressions.addAll(searchAggregationExpressions(expression));
+ }
+ if (queryStatement.hasHaving()) {
+ aggregationExpressions.addAll(searchAggregationExpressions(analysis.getHavingExpression()));
+ }
+ if (queryStatement.hasOrderByExpression()) {
+ for (Expression expression : analysis.getOrderByExpressions()) {
+ aggregationExpressions.addAll(searchAggregationExpressions(expression));
}
- analysis.setAggregationExpressions(aggregationExpressions);
}
+ analysis.setAggregationExpressions(aggregationExpressions);
}
private void analyzeDeviceToSourceTransform(Analysis analysis, QueryStatement queryStatement) {
if (queryStatement.isAggregationQuery()) {
- Map<String, Set<Expression>> deviceToSourceTransformExpressions = new HashMap<>();
-
+ Map<String, Set<Expression>> deviceToSourceTransformExpressions =
+ analysis.getDeviceToSourceTransformExpressions();
Map<String, Set<Expression>> deviceToAggregationExpressions =
analysis.getDeviceToAggregationExpressions();
- for (Map.Entry<String, Set<Expression>> deviceAggregationExpressionsEntry :
- deviceToAggregationExpressions.entrySet()) {
- String deviceName = deviceAggregationExpressionsEntry.getKey();
- Set<Expression> aggregationExpressions = deviceAggregationExpressionsEntry.getValue();
- Set<Expression> sourceTransformExpressions = new LinkedHashSet<>();
+ for (Map.Entry<String, Set<Expression>> entry : deviceToAggregationExpressions.entrySet()) {
+ String deviceName = entry.getKey();
+ Set<Expression> aggregationExpressions = entry.getValue();
+
+ Set<Expression> sourceTransformExpressions =
+ deviceToSourceTransformExpressions.computeIfAbsent(
+ deviceName, k -> new LinkedHashSet<>());
+
for (Expression expression : aggregationExpressions) {
- // We just process first input Expression of AggregationFunction,
- // keep other input Expressions as origin
- // If AggregationFunction need more than one input series,
- // we need to reconsider the process of it
- sourceTransformExpressions.add(expression.getExpressions().get(0));
+ // if count_time aggregation exist, it can exist only one count_time(*)
+ if (queryStatement.isCountTimeAggregation()) {
+ for (Expression countTimeSourceExpression :
+ ((FunctionExpression) expression).getCountTimeExpressions()) {
+
+ analyzeExpressionType(analysis, countTimeSourceExpression);
+ sourceTransformExpressions.add(countTimeSourceExpression);
+ }
+ } else {
+ // We just process first input Expression of AggregationFunction,
+ // keep other input Expressions as origin
+ // If AggregationFunction need more than one input series,
+ // we need to reconsider the process of it
+ sourceTransformExpressions.add(expression.getExpressions().get(0));
+ }
}
+
if (queryStatement.hasGroupByExpression()) {
sourceTransformExpressions.add(analysis.getDeviceToGroupByExpression().get(deviceName));
}
- deviceToSourceTransformExpressions.put(deviceName, sourceTransformExpressions);
}
- analysis.setDeviceToSourceTransformExpressions(deviceToSourceTransformExpressions);
} else {
updateDeviceToSourceTransformAndOutputExpressions(
analysis, analysis.getDeviceToSelectExpressions());
@@ -1099,7 +1122,7 @@ public class AnalyzeVisitor extends StatementVisitor<Analysis, MPPQueryContext>
Set<Expression> normalizedExpressions = new LinkedHashSet<>();
for (Expression expression : expressions) {
- Expression normalizedExpression = ExpressionAnalyzer.normalizeExpression(expression);
+ Expression normalizedExpression = normalizeExpression(expression);
analyzeExpressionType(analysis, normalizedExpression);
normalizedExpressions.add(normalizedExpression);
@@ -1113,13 +1136,36 @@ public class AnalyzeVisitor extends StatementVisitor<Analysis, MPPQueryContext>
}
}
- private void analyzeSourceTransform(Analysis analysis, QueryStatement queryStatement) {
- Set<Expression> sourceTransformExpressions = new HashSet<>();
+ private void analyzeSourceTransform(
+ Analysis analysis,
+ List<Pair<Expression, String>> outputExpressions,
+ QueryStatement queryStatement) {
+ Set<Expression> sourceTransformExpressions = analysis.getSourceTransformExpressions();
+
if (queryStatement.isAggregationQuery()) {
- for (Expression expression : analysis.getAggregationExpressions()) {
- // for AggregationExpression, only the first Expression of input need to transform
- sourceTransformExpressions.add(expression.getExpressions().get(0));
+ if (queryStatement.isCountTimeAggregation()) {
+
+ for (Pair<Expression, String> pair : outputExpressions) {
+ FunctionExpression countTimeExpression = (FunctionExpression) pair.left;
+ for (Expression countTimeSourceExpression :
+ countTimeExpression.getCountTimeExpressions()) {
+ analyzeExpressionType(analysis, countTimeSourceExpression);
+ sourceTransformExpressions.add(countTimeSourceExpression);
+ }
+ }
+
+ // count_time only returns one result
+ Pair<Expression, String> firstCountTimeExpression = outputExpressions.get(0);
+ outputExpressions.clear();
+ outputExpressions.add(firstCountTimeExpression);
+
+ } else {
+ for (Expression aggExpression : analysis.getAggregationExpressions()) {
+ // for AggregationExpression, only the first Expression of input need to transform
+ sourceTransformExpressions.add(aggExpression.getExpressions().get(0));
+ }
}
+
if (queryStatement.hasGroupByExpression()) {
sourceTransformExpressions.add(analysis.getGroupByExpression());
}
@@ -1129,24 +1175,24 @@ public class AnalyzeVisitor extends StatementVisitor<Analysis, MPPQueryContext>
sourceTransformExpressions.addAll(analysis.getOrderByExpressions());
}
}
- analysis.setSourceTransformExpressions(sourceTransformExpressions);
}
private void analyzeDeviceToSource(Analysis analysis, QueryStatement queryStatement) {
Map<String, Set<Expression>> deviceToSourceExpressions = new HashMap<>();
Map<String, Set<Expression>> deviceToSourceTransformExpressions =
analysis.getDeviceToSourceTransformExpressions();
- for (Map.Entry<String, Set<Expression>> deviceSourceTransformExpressionsEntry :
- deviceToSourceTransformExpressions.entrySet()) {
- String deviceName = deviceSourceTransformExpressionsEntry.getKey();
- Set<Expression> sourceTransformExpressions = deviceSourceTransformExpressionsEntry.getValue();
+
+ for (Map.Entry<String, Set<Expression>> entry : deviceToSourceTransformExpressions.entrySet()) {
+ String deviceName = entry.getKey();
+ Set<Expression> sourceTransformExpressions = entry.getValue();
Set<Expression> sourceExpressions = new LinkedHashSet<>();
- for (Expression expression : sourceTransformExpressions) {
- sourceExpressions.addAll(ExpressionAnalyzer.searchSourceExpressions(expression));
- }
+ sourceTransformExpressions.forEach(
+ expression -> sourceExpressions.addAll(searchSourceExpressions(expression)));
+
deviceToSourceExpressions.put(deviceName, sourceExpressions);
}
+
if (queryStatement.hasWhere()) {
Map<String, Expression> deviceToWhereExpression = analysis.getDeviceToWhereExpression();
for (Map.Entry<String, Expression> deviceWhereExpressionEntry :
@@ -1155,14 +1201,14 @@ public class AnalyzeVisitor extends StatementVisitor<Analysis, MPPQueryContext>
Expression whereExpression = deviceWhereExpressionEntry.getValue();
deviceToSourceExpressions
.computeIfAbsent(deviceName, key -> new LinkedHashSet<>())
- .addAll(ExpressionAnalyzer.searchSourceExpressions(whereExpression));
+ .addAll(searchSourceExpressions(whereExpression));
}
}
Map<String, List<String>> outputDeviceToQueriedDevicesMap = new LinkedHashMap<>();
- for (Map.Entry<String, Set<Expression>> deviceSourceExpressionsEntry :
- deviceToSourceExpressions.entrySet()) {
- Set<Expression> sourceExpressionsUnderDevice = deviceSourceExpressionsEntry.getValue();
+ for (Map.Entry<String, Set<Expression>> entry : deviceToSourceExpressions.entrySet()) {
+ String deviceName = entry.getKey();
+ Set<Expression> sourceExpressionsUnderDevice = entry.getValue();
Set<String> queriedDevices = new HashSet<>();
for (Expression expression : sourceExpressionsUnderDevice) {
queriedDevices.add(ExpressionAnalyzer.getDeviceNameInSourceExpression(expression));
@@ -1171,8 +1217,7 @@ public class AnalyzeVisitor extends StatementVisitor<Analysis, MPPQueryContext>
throw new SemanticException(
"Cross-device queries are not supported in ALIGN BY DEVICE queries.");
}
- outputDeviceToQueriedDevicesMap.put(
- deviceSourceExpressionsEntry.getKey(), new ArrayList<>(queriedDevices));
+ outputDeviceToQueriedDevicesMap.put(deviceName, new ArrayList<>(queriedDevices));
}
analysis.setDeviceToSourceExpressions(deviceToSourceExpressions);
@@ -1180,15 +1225,18 @@ public class AnalyzeVisitor extends StatementVisitor<Analysis, MPPQueryContext>
}
private void analyzeSource(Analysis analysis, QueryStatement queryStatement) {
- Set<Expression> sourceExpressions = new HashSet<>();
+ Set<Expression> sourceExpressions = analysis.getSourceExpressions();
+ if (sourceExpressions == null) {
+ sourceExpressions = new HashSet<>();
+ analysis.setSourceExpressions(sourceExpressions);
+ }
+
for (Expression expression : analysis.getSourceTransformExpressions()) {
- sourceExpressions.addAll(ExpressionAnalyzer.searchSourceExpressions(expression));
+ sourceExpressions.addAll(searchSourceExpressions(expression));
}
if (queryStatement.hasWhere()) {
- sourceExpressions.addAll(
- ExpressionAnalyzer.searchSourceExpressions(analysis.getWhereExpression()));
+ sourceExpressions.addAll(searchSourceExpressions(analysis.getWhereExpression()));
}
- analysis.setSourceExpressions(sourceExpressions);
}
private static final String WHERE_WRONG_TYPE_ERROR_MSG =
@@ -1210,8 +1258,7 @@ public class AnalyzeVisitor extends StatementVisitor<Analysis, MPPQueryContext>
Expression whereExpression;
try {
whereExpression =
- ExpressionAnalyzer.normalizeExpression(
- analyzeWhereSplitByDevice(queryStatement, devicePath, schemaTree));
+ normalizeExpression(analyzeWhereSplitByDevice(queryStatement, devicePath, schemaTree));
} catch (MeasurementNotExistException e) {
logger.warn(
"Meets MeasurementNotExistException in analyzeDeviceToWhere when executing align by device, "
@@ -1245,7 +1292,7 @@ public class AnalyzeVisitor extends StatementVisitor<Analysis, MPPQueryContext>
Expression whereExpression =
ExpressionUtils.constructQueryFilter(
conJunctions.stream().distinct().collect(Collectors.toList()));
- whereExpression = ExpressionAnalyzer.normalizeExpression(whereExpression);
+ whereExpression = normalizeExpression(whereExpression);
TSDataType outputType = analyzeExpressionType(analysis, whereExpression);
if (outputType != TSDataType.BOOLEAN) {
throw new SemanticException(String.format(WHERE_WRONG_TYPE_ERROR_MSG, outputType));
@@ -1266,22 +1313,20 @@ public class AnalyzeVisitor extends StatementVisitor<Analysis, MPPQueryContext>
Set<Expression> selectExpressions = analysis.getSelectExpressions();
Set<Expression> deviceViewOutputExpressions = new LinkedHashSet<>();
if (queryStatement.isAggregationQuery()) {
- deviceViewOutputExpressions.add(deviceExpression);
+ deviceViewOutputExpressions.add(DEVICE_EXPRESSION);
if (queryStatement.isOutputEndTime()) {
- deviceViewOutputExpressions.add(endTimeExpression);
+ deviceViewOutputExpressions.add(END_TIME_EXPRESSION);
}
for (Expression selectExpression : selectExpressions) {
- deviceViewOutputExpressions.addAll(
- ExpressionAnalyzer.searchAggregationExpressions(selectExpression));
+ deviceViewOutputExpressions.addAll(searchAggregationExpressions(selectExpression));
}
if (queryStatement.hasHaving()) {
deviceViewOutputExpressions.addAll(
- ExpressionAnalyzer.searchAggregationExpressions(analysis.getHavingExpression()));
+ searchAggregationExpressions(analysis.getHavingExpression()));
}
if (queryStatement.hasOrderByExpression()) {
for (Expression orderByExpression : analysis.getOrderByExpressions()) {
- deviceViewOutputExpressions.addAll(
- ExpressionAnalyzer.searchAggregationExpressions(orderByExpression));
+ deviceViewOutputExpressions.addAll(searchAggregationExpressions(orderByExpression));
}
}
} else {
@@ -1332,8 +1377,7 @@ public class AnalyzeVisitor extends StatementVisitor<Analysis, MPPQueryContext>
outputColumns.add(ENDTIME);
}
for (Expression expression : outputExpressionsUnderDevice) {
- outputColumns.add(
- ExpressionAnalyzer.getMeasurementExpression(expression, analysis).getOutputSymbol());
+ outputColumns.add(getMeasurementExpression(expression, analysis).getOutputSymbol());
}
deviceToOutputColumnsMap.put(deviceOutputExpressionEntry.getKey(), outputColumns);
}
@@ -1422,8 +1466,7 @@ public class AnalyzeVisitor extends StatementVisitor<Analysis, MPPQueryContext>
Set<Expression> orderByExpressions = new LinkedHashSet<>();
for (Expression expressionForItem : queryStatement.getExpressionSortItemList()) {
// Expression in a sortItem only indicates one column
- List<Expression> expressions =
- ExpressionAnalyzer.bindSchemaForExpression(expressionForItem, schemaTree);
+ List<Expression> expressions = bindSchemaForExpression(expressionForItem, schemaTree);
if (expressions.isEmpty()) {
throw new SemanticException(
String.format(
@@ -1435,7 +1478,7 @@ public class AnalyzeVisitor extends StatementVisitor<Analysis, MPPQueryContext>
"%s in order by clause shouldn't refer to more than one timeseries.",
expressionForItem.getExpressionString()));
}
- expressionForItem = ExpressionAnalyzer.normalizeExpression(expressions.get(0));
+ expressionForItem = normalizeExpression(expressions.get(0));
TSDataType dataType = analyzeExpressionType(analysis, expressionForItem);
if (!dataType.isComparable()) {
throw new SemanticException(
@@ -1467,8 +1510,7 @@ public class AnalyzeVisitor extends StatementVisitor<Analysis, MPPQueryContext>
Expression expression = groupByComponent.getControlColumnExpression();
for (PartialPath device : deviceSet) {
List<Expression> groupByExpressionsOfOneDevice =
- ExpressionAnalyzer.concatDeviceAndBindSchemaForExpression(
- expression, device, schemaTree);
+ concatDeviceAndBindSchemaForExpression(expression, device, schemaTree);
if (groupByExpressionsOfOneDevice.isEmpty()) {
throw new SemanticException(
@@ -1481,8 +1523,7 @@ public class AnalyzeVisitor extends StatementVisitor<Analysis, MPPQueryContext>
expression));
}
deviceToGroupByExpression.put(
- device.getFullPath(),
- ExpressionAnalyzer.normalizeExpression(groupByExpressionsOfOneDevice.get(0)));
+ device.getFullPath(), normalizeExpression(groupByExpressionsOfOneDevice.get(0)));
}
}
@@ -1541,8 +1582,7 @@ public class AnalyzeVisitor extends StatementVisitor<Analysis, MPPQueryContext>
Set<Expression> orderByExpressionsForOneDevice = new LinkedHashSet<>();
for (Expression expressionForItem : queryStatement.getExpressionSortItemList()) {
List<Expression> expressions =
- ExpressionAnalyzer.concatDeviceAndBindSchemaForExpression(
- expressionForItem, device, schemaTree);
+ concatDeviceAndBindSchemaForExpression(expressionForItem, device, schemaTree);
if (expressions.isEmpty()) {
throw new SemanticException(
String.format(
@@ -1561,8 +1601,7 @@ public class AnalyzeVisitor extends StatementVisitor<Analysis, MPPQueryContext>
String.format("The data type of %s is not comparable", dataType));
}
- Expression deviceViewExpression =
- ExpressionAnalyzer.getMeasurementExpression(expressionForItem, analysis);
+ Expression deviceViewExpression = getMeasurementExpression(expressionForItem, analysis);
analyzeExpressionType(analysis, deviceViewExpression);
deviceViewOrderByExpression.add(deviceViewExpression);
@@ -1592,8 +1631,7 @@ public class AnalyzeVisitor extends StatementVisitor<Analysis, MPPQueryContext>
if (queryStatement.hasGroupByExpression()) {
groupByExpression = groupByComponent.getControlColumnExpression();
// Expression in group by variation clause only indicates one column
- List<Expression> expressions =
- ExpressionAnalyzer.bindSchemaForExpression(groupByExpression, schemaTree);
+ List<Expression> expressions = bindSchemaForExpression(groupByExpression, schemaTree);
if (expressions.isEmpty()) {
throw new SemanticException(
String.format(
@@ -1606,12 +1644,11 @@ public class AnalyzeVisitor extends StatementVisitor<Analysis, MPPQueryContext>
groupByExpression.getExpressionString()));
}
// Aggregation expression shouldn't exist in group by clause.
- List<Expression> aggregationExpression =
- ExpressionAnalyzer.searchAggregationExpressions(expressions.get(0));
+ List<Expression> aggregationExpression = searchAggregationExpressions(expressions.get(0));
if (aggregationExpression != null && !aggregationExpression.isEmpty()) {
throw new SemanticException("Aggregation expression shouldn't exist in group by clause");
}
- groupByExpression = ExpressionAnalyzer.normalizeExpression(expressions.get(0));
+ groupByExpression = normalizeExpression(expressions.get(0));
}
if (windowType == WindowType.VARIATION_WINDOW) {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/ExpressionAnalyzer.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/ExpressionAnalyzer.java
index 886d284edd2..28b86e4a3d4 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/ExpressionAnalyzer.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/ExpressionAnalyzer.java
@@ -698,6 +698,8 @@ public class ExpressionAnalyzer {
return true;
} else if (expression instanceof CaseWhenThenExpression) {
return true;
+ } else if (expression instanceof TimestampOperand) {
+ return false;
} else {
throw new UnknownExpressionTypeException(expression.getExpressionType());
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/ExpressionTypeAnalyzer.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/ExpressionTypeAnalyzer.java
index e52fb95f481..41cd1fbe882 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/ExpressionTypeAnalyzer.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/ExpressionTypeAnalyzer.java
@@ -63,7 +63,7 @@ public class ExpressionTypeAnalyzer {
ExpressionTypeAnalyzer analyzer = new ExpressionTypeAnalyzer();
analyzer.analyze(expression);
- updateAnalysis(analysis, analyzer);
+ addExpressionTypes(analysis, analyzer);
}
return analysis.getType(expression);
}
@@ -76,7 +76,7 @@ public class ExpressionTypeAnalyzer {
types.putAll(analyzer.getExpressionTypes());
}
- private static void updateAnalysis(Analysis analysis, ExpressionTypeAnalyzer analyzer) {
+ private static void addExpressionTypes(Analysis analysis, ExpressionTypeAnalyzer analyzer) {
analysis.addTypes(analyzer.getExpressionTypes());
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/ExpressionUtils.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/ExpressionUtils.java
index 28d5a7b2dd3..69533a3ff4c 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/ExpressionUtils.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/ExpressionUtils.java
@@ -97,7 +97,8 @@ public class ExpressionUtils {
new FunctionExpression(
rawExpression.getFunctionName(),
rawExpression.getFunctionAttributes(),
- childExpressions);
+ childExpressions,
+ rawExpression.getCountTimeExpressions());
return cloneCommonFields(rawExpression, resultExpression);
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/expression/ExpressionFactory.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/expression/ExpressionFactory.java
index d083fe43bec..741e7c4ccc0 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/expression/ExpressionFactory.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/expression/ExpressionFactory.java
@@ -89,6 +89,13 @@ public class ExpressionFactory {
Collections.singletonList(inputExpression));
}
+ public static FunctionExpression countTime(Expression inputExpression) {
+ return new FunctionExpression(
+ TAggregationType.COUNT_TIME.toString(),
+ new LinkedHashMap<>(),
+ Collections.singletonList(inputExpression));
+ }
+
public static FunctionExpression sum(Expression inputExpression) {
return new FunctionExpression(
TAggregationType.SUM.toString(),
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/expression/binary/BinaryExpression.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/expression/binary/BinaryExpression.java
index e967cc7e4d3..2f5637ddadc 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/expression/binary/BinaryExpression.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/expression/binary/BinaryExpression.java
@@ -152,6 +152,10 @@ public abstract class BinaryExpression extends Expression {
String left = this.getLeftExpression().getOutputSymbol();
String right = this.getRightExpression().getOutputSymbol();
+ return buildExpression(left, right);
+ }
+
+ private String buildExpression(String left, String right) {
StringBuilder builder = new StringBuilder();
if (leftExpression.getExpressionType().getPriority() < this.getExpressionType().getPriority()) {
builder.append("(").append(left).append(")");
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/expression/leaf/TimestampOperand.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/expression/leaf/TimestampOperand.java
index 3b9e72d5dcf..ece2f5ad9ef 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/expression/leaf/TimestampOperand.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/expression/leaf/TimestampOperand.java
@@ -19,7 +19,6 @@
package org.apache.iotdb.db.queryengine.plan.expression.leaf;
-import org.apache.iotdb.commons.path.PartialPath;
import org.apache.iotdb.db.queryengine.plan.expression.ExpressionType;
import org.apache.iotdb.db.queryengine.plan.expression.visitor.ExpressionVisitor;
import org.apache.iotdb.db.queryengine.plan.planner.plan.parameter.InputLocation;
@@ -33,7 +32,7 @@ import java.util.Map;
public class TimestampOperand extends LeafOperand {
- public static final PartialPath TIMESTAMP_PARTIAL_PATH = new PartialPath("Time", false);
+ public static final String TIMESTAMP_EXPRESSION_STRING = "Time";
public TimestampOperand() {
// do nothing
@@ -66,7 +65,7 @@ public class TimestampOperand extends LeafOperand {
@Override
protected String getExpressionStringInternal() {
- return "Time";
+ return TIMESTAMP_EXPRESSION_STRING;
}
@Override
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/expression/multi/FunctionExpression.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/expression/multi/FunctionExpression.java
index 5807b4836b9..17b628f1cb5 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/expression/multi/FunctionExpression.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/expression/multi/FunctionExpression.java
@@ -65,6 +65,8 @@ public class FunctionExpression extends Expression {
private List<PartialPath> paths;
+ private List<Expression> countTimeExpressions;
+
private String parametersString;
public FunctionExpression(String functionName) {
@@ -80,6 +82,18 @@ public class FunctionExpression extends Expression {
this.functionName = functionName;
this.functionAttributes = functionAttributes;
this.expressions = expressions;
+ this.countTimeExpressions = null;
+ }
+
+ public FunctionExpression(
+ String functionName,
+ LinkedHashMap<String, String> functionAttributes,
+ List<Expression> expressions,
+ List<Expression> countTimeExpressions) {
+ this.functionName = functionName;
+ this.functionAttributes = functionAttributes;
+ this.expressions = expressions;
+ this.countTimeExpressions = countTimeExpressions;
}
public FunctionExpression(ByteBuffer byteBuffer) {
@@ -276,6 +290,10 @@ public class FunctionExpression extends Expression {
return paths;
}
+ public List<Expression> getCountTimeExpressions() {
+ return this.countTimeExpressions;
+ }
+
@Override
public String getExpressionStringInternal() {
return functionName + "(" + getParametersString() + ")";
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/expression/visitor/LowercaseNormalizeVisitor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/expression/visitor/CountTimeAggregationAmountVisitor.java
similarity index 50%
copy from iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/expression/visitor/LowercaseNormalizeVisitor.java
copy to iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/expression/visitor/CountTimeAggregationAmountVisitor.java
index bcf97db91f6..1a573f856c2 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/expression/visitor/LowercaseNormalizeVisitor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/expression/visitor/CountTimeAggregationAmountVisitor.java
@@ -19,21 +19,36 @@
package org.apache.iotdb.db.queryengine.plan.expression.visitor;
+import org.apache.iotdb.db.exception.sql.SemanticException;
import org.apache.iotdb.db.queryengine.plan.expression.Expression;
+import org.apache.iotdb.db.queryengine.plan.expression.leaf.LeafOperand;
import org.apache.iotdb.db.queryengine.plan.expression.multi.FunctionExpression;
-import java.util.ArrayList;
+import java.util.Collections;
import java.util.List;
-public class LowercaseNormalizeVisitor extends ReconstructVisitor<Void> {
+import static org.apache.iotdb.db.utils.constant.SqlConstant.COUNT_TIME;
+
+public class CountTimeAggregationAmountVisitor extends CollectVisitor {
+
+ public static final String COUNT_TIME_ONLY_SUPPORT_ONE_WILDCARD =
+ "The parameter of count_time aggregation can only be '*'.";
+
+ @Override
+ public List<Expression> visitFunctionExpression(
+ FunctionExpression functionExpression, Void context) {
+ if (COUNT_TIME.equalsIgnoreCase(functionExpression.getFunctionName())) {
+ if (!"*".equals(functionExpression.getExpressions().get(0).toString())) {
+ throw new SemanticException(COUNT_TIME_ONLY_SUPPORT_ONE_WILDCARD);
+ }
+ return Collections.singletonList(functionExpression);
+ }
+
+ return Collections.emptyList();
+ }
@Override
- public Expression visitFunctionExpression(FunctionExpression functionExpression, Void context) {
- List<Expression> childResult = new ArrayList<>();
- functionExpression.getExpressions().forEach(child -> childResult.add(process(child, null)));
- return new FunctionExpression(
- functionExpression.getFunctionName().toLowerCase(),
- functionExpression.getFunctionAttributes(),
- childResult);
+ public List<Expression> visitLeafOperand(LeafOperand leafOperand, Void context) {
+ return Collections.emptyList();
}
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/expression/visitor/ExpressionNormalizeVisitor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/expression/visitor/ExpressionNormalizeVisitor.java
index fdec15b17d5..ab6212b637e 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/expression/visitor/ExpressionNormalizeVisitor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/expression/visitor/ExpressionNormalizeVisitor.java
@@ -44,7 +44,8 @@ public class ExpressionNormalizeVisitor extends ReconstructVisitor<Void> {
return new FunctionExpression(
functionExpression.getFunctionName().toLowerCase(),
functionExpression.getFunctionAttributes(),
- childResult);
+ childResult,
+ functionExpression.getCountTimeExpressions());
}
@Override
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/expression/visitor/GetMeasurementExpressionVisitor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/expression/visitor/GetMeasurementExpressionVisitor.java
index 9ce4d9ccb0b..2dc87a9fe78 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/expression/visitor/GetMeasurementExpressionVisitor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/expression/visitor/GetMeasurementExpressionVisitor.java
@@ -54,7 +54,8 @@ public class GetMeasurementExpressionVisitor extends ReconstructVisitor<Analysis
return new FunctionExpression(
functionExpression.getFunctionName(),
functionExpression.getFunctionAttributes(),
- childExpressions);
+ childExpressions,
+ functionExpression.getCountTimeExpressions());
}
@Override
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/expression/visitor/LowercaseNormalizeVisitor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/expression/visitor/LowercaseNormalizeVisitor.java
index bcf97db91f6..854a56e937a 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/expression/visitor/LowercaseNormalizeVisitor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/expression/visitor/LowercaseNormalizeVisitor.java
@@ -34,6 +34,7 @@ public class LowercaseNormalizeVisitor extends ReconstructVisitor<Void> {
return new FunctionExpression(
functionExpression.getFunctionName().toLowerCase(),
functionExpression.getFunctionAttributes(),
- childResult);
+ childResult,
+ functionExpression.getCountTimeExpressions());
}
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/expression/visitor/cartesian/BindSchemaForExpressionVisitor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/expression/visitor/cartesian/BindSchemaForExpressionVisitor.java
index 4c49886e509..eb036e7799b 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/expression/visitor/cartesian/BindSchemaForExpressionVisitor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/expression/visitor/cartesian/BindSchemaForExpressionVisitor.java
@@ -36,17 +36,36 @@ import org.apache.iotdb.tsfile.write.schema.IMeasurementSchema;
import java.util.ArrayList;
import java.util.Collections;
+import java.util.LinkedHashMap;
import java.util.List;
+import java.util.stream.Collectors;
import static org.apache.iotdb.db.queryengine.plan.analyze.ExpressionUtils.cartesianProduct;
import static org.apache.iotdb.db.queryengine.plan.analyze.ExpressionUtils.reconstructFunctionExpressions;
import static org.apache.iotdb.db.utils.TypeInferenceUtils.bindTypeForAggregationNonSeriesInputExpressions;
+import static org.apache.iotdb.db.utils.constant.SqlConstant.COUNT_TIME;
public class BindSchemaForExpressionVisitor extends CartesianProductVisitor<ISchemaTree> {
@Override
public List<Expression> visitFunctionExpression(
FunctionExpression functionExpression, ISchemaTree schemaTree) {
+
+ if (COUNT_TIME.equalsIgnoreCase(functionExpression.getFunctionName())) {
+ List<Expression> usedExpressions =
+ functionExpression.getExpressions().stream()
+ .flatMap(e -> process(e, schemaTree).stream())
+ .collect(Collectors.toList());
+
+ Expression countTimeExpression =
+ new FunctionExpression(
+ COUNT_TIME,
+ new LinkedHashMap<>(),
+ Collections.singletonList(new TimestampOperand()),
+ usedExpressions);
+ return Collections.singletonList(countTimeExpression);
+ }
+
// One by one, remove the wildcards from the input expressions. In most cases, an expression
// will produce multiple expressions after removing the wildcards. We use extendedExpressions
// to collect the produced expressions.
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/expression/visitor/cartesian/BindSchemaForPredicateVisitor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/expression/visitor/cartesian/BindSchemaForPredicateVisitor.java
index 4320ad309f4..bca177d4505 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/expression/visitor/cartesian/BindSchemaForPredicateVisitor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/expression/visitor/cartesian/BindSchemaForPredicateVisitor.java
@@ -34,7 +34,9 @@ import org.apache.iotdb.db.utils.constant.SqlConstant;
import java.util.ArrayList;
import java.util.Collections;
+import java.util.LinkedHashMap;
import java.util.List;
+import java.util.stream.Collectors;
import static org.apache.iotdb.db.queryengine.plan.analyze.ExpressionUtils.cartesianProduct;
import static org.apache.iotdb.db.queryengine.plan.analyze.ExpressionUtils.reconstructBinaryExpressions;
@@ -42,6 +44,7 @@ import static org.apache.iotdb.db.queryengine.plan.analyze.ExpressionUtils.recon
import static org.apache.iotdb.db.queryengine.plan.analyze.ExpressionUtils.reconstructTimeSeriesOperands;
import static org.apache.iotdb.db.queryengine.plan.expression.visitor.cartesian.BindSchemaForExpressionVisitor.transformViewPath;
import static org.apache.iotdb.db.utils.TypeInferenceUtils.bindTypeForAggregationNonSeriesInputExpressions;
+import static org.apache.iotdb.db.utils.constant.SqlConstant.COUNT_TIME;
public class BindSchemaForPredicateVisitor
extends CartesianProductVisitor<BindSchemaForPredicateVisitor.Context> {
@@ -63,6 +66,21 @@ public class BindSchemaForPredicateVisitor
@Override
public List<Expression> visitFunctionExpression(FunctionExpression predicate, Context context) {
+ if (COUNT_TIME.equalsIgnoreCase(predicate.getFunctionName())) {
+ List<Expression> usedExpressions =
+ predicate.getExpressions().stream()
+ .flatMap(e -> process(e, context).stream())
+ .collect(Collectors.toList());
+
+ Expression countTimeExpression =
+ new FunctionExpression(
+ COUNT_TIME,
+ new LinkedHashMap<>(),
+ Collections.singletonList(new TimestampOperand()),
+ usedExpressions);
+ return Collections.singletonList(countTimeExpression);
+ }
+
List<List<Expression>> extendedExpressions = new ArrayList<>();
for (Expression suffixExpression : predicate.getExpressions()) {
extendedExpressions.add(
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/expression/visitor/cartesian/ConcatDeviceAndBindSchemaForExpressionVisitor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/expression/visitor/cartesian/ConcatDeviceAndBindSchemaForExpressionVisitor.java
index 9292204dc82..8217d0a92fd 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/expression/visitor/cartesian/ConcatDeviceAndBindSchemaForExpressionVisitor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/expression/visitor/cartesian/ConcatDeviceAndBindSchemaForExpressionVisitor.java
@@ -31,13 +31,17 @@ import org.apache.iotdb.db.queryengine.plan.expression.leaf.TimestampOperand;
import org.apache.iotdb.db.queryengine.plan.expression.multi.FunctionExpression;
import java.util.ArrayList;
+import java.util.Collection;
import java.util.Collections;
+import java.util.LinkedHashMap;
import java.util.List;
+import java.util.stream.Collectors;
import static org.apache.iotdb.db.queryengine.plan.analyze.ExpressionUtils.cartesianProduct;
import static org.apache.iotdb.db.queryengine.plan.analyze.ExpressionUtils.reconstructFunctionExpressions;
import static org.apache.iotdb.db.queryengine.plan.expression.visitor.cartesian.BindSchemaForExpressionVisitor.transformViewPath;
import static org.apache.iotdb.db.utils.TypeInferenceUtils.bindTypeForAggregationNonSeriesInputExpressions;
+import static org.apache.iotdb.db.utils.constant.SqlConstant.COUNT_TIME;
public class ConcatDeviceAndBindSchemaForExpressionVisitor
extends CartesianProductVisitor<ConcatDeviceAndBindSchemaForExpressionVisitor.Context> {
@@ -47,9 +51,9 @@ public class ConcatDeviceAndBindSchemaForExpressionVisitor
FunctionExpression functionExpression, Context context) {
List<List<Expression>> extendedExpressions = new ArrayList<>();
for (Expression suffixExpression : functionExpression.getExpressions()) {
- List<Expression> concatedExpression = process(suffixExpression, context);
- if (concatedExpression != null && !concatedExpression.isEmpty()) {
- extendedExpressions.add(concatedExpression);
+ List<Expression> concatExpression = process(suffixExpression, context);
+ if (concatExpression != null && !concatExpression.isEmpty()) {
+ extendedExpressions.add(concatExpression);
}
// We just process first input Expression of AggregationFunction,
@@ -64,6 +68,19 @@ public class ConcatDeviceAndBindSchemaForExpressionVisitor
}
}
+ if (COUNT_TIME.equalsIgnoreCase(functionExpression.getFunctionName())) {
+ List<Expression> usedExpressions =
+ extendedExpressions.stream().flatMap(Collection::stream).collect(Collectors.toList());
+
+ Expression countTimeExpression =
+ new FunctionExpression(
+ COUNT_TIME,
+ new LinkedHashMap<>(),
+ Collections.singletonList(new TimestampOperand()),
+ usedExpressions);
+ return Collections.singletonList(countTimeExpression);
+ }
+
List<List<Expression>> childExpressionsList = new ArrayList<>();
cartesianProduct(extendedExpressions, childExpressionsList, 0, new ArrayList<>());
return reconstructFunctionExpressions(functionExpression, childExpressionsList);
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/parser/ASTVisitor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/parser/ASTVisitor.java
index 72ab30672b3..c8299c50a68 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/parser/ASTVisitor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/parser/ASTVisitor.java
@@ -2857,6 +2857,7 @@ public class ASTVisitor extends IoTDBSqlParserBaseVisitor<Statement> {
case SqlConstant.MIN_TIME:
case SqlConstant.MAX_TIME:
case SqlConstant.COUNT:
+ case SqlConstant.COUNT_TIME:
case SqlConstant.MIN_VALUE:
case SqlConstant.LAST_VALUE:
case SqlConstant.FIRST_VALUE:
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/LogicalPlanBuilder.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/LogicalPlanBuilder.java
index 31684bf0915..54f2ce71dbc 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/LogicalPlanBuilder.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/LogicalPlanBuilder.java
@@ -23,6 +23,7 @@ import org.apache.iotdb.common.rpc.thrift.TAggregationType;
import org.apache.iotdb.common.rpc.thrift.TDataNodeLocation;
import org.apache.iotdb.common.rpc.thrift.TSchemaNode;
import org.apache.iotdb.commons.exception.IllegalPathException;
+import org.apache.iotdb.commons.exception.MetadataException;
import org.apache.iotdb.commons.path.AlignedPath;
import org.apache.iotdb.commons.path.MeasurementPath;
import org.apache.iotdb.commons.path.PartialPath;
@@ -97,6 +98,8 @@ import org.apache.iotdb.db.schemaengine.template.Template;
import org.apache.iotdb.db.utils.SchemaUtils;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.read.filter.basic.Filter;
+import org.apache.iotdb.tsfile.utils.Pair;
+import org.apache.iotdb.tsfile.write.schema.IMeasurementSchema;
import com.google.common.base.Function;
import org.apache.commons.lang3.Validate;
@@ -118,6 +121,7 @@ import static com.google.common.base.Preconditions.checkArgument;
import static org.apache.iotdb.commons.conf.IoTDBConstant.MULTI_LEVEL_PATH_WILDCARD;
import static org.apache.iotdb.db.queryengine.common.header.ColumnHeaderConstant.DEVICE;
import static org.apache.iotdb.db.queryengine.common.header.ColumnHeaderConstant.ENDTIME;
+import static org.apache.iotdb.db.utils.constant.SqlConstant.COUNT_TIME;
public class LogicalPlanBuilder {
@@ -324,6 +328,7 @@ public class LogicalPlanBuilder {
boolean needCheckAscending = groupByTimeParameter == null;
Map<PartialPath, List<AggregationDescriptor>> ascendingAggregations = new HashMap<>();
Map<PartialPath, List<AggregationDescriptor>> descendingAggregations = new HashMap<>();
+ Map<PartialPath, List<AggregationDescriptor>> countTimeAggregations = new HashMap<>();
for (Expression aggregationExpression : aggregationExpressions) {
createAggregationDescriptor(
(FunctionExpression) aggregationExpression,
@@ -331,13 +336,15 @@ public class LogicalPlanBuilder {
scanOrder,
needCheckAscending,
ascendingAggregations,
- descendingAggregations);
+ descendingAggregations,
+ countTimeAggregations);
}
List<PlanNode> sourceNodeList =
constructSourceNodeFromAggregationDescriptors(
ascendingAggregations,
descendingAggregations,
+ countTimeAggregations,
scanOrder,
timeFilter,
groupByTimeParameter);
@@ -372,6 +379,7 @@ public class LogicalPlanBuilder {
Map<PartialPath, List<AggregationDescriptor>> ascendingAggregations = new HashMap<>();
Map<PartialPath, List<AggregationDescriptor>> descendingAggregations = new HashMap<>();
Map<AggregationDescriptor, Integer> aggregationToIndexMap = new HashMap<>();
+ Map<PartialPath, List<AggregationDescriptor>> countTimeAggregations = new HashMap<>();
int index = 0;
for (Expression aggregationExpression : aggregationExpressions) {
@@ -382,7 +390,8 @@ public class LogicalPlanBuilder {
scanOrder,
needCheckAscending,
ascendingAggregations,
- descendingAggregations);
+ descendingAggregations,
+ countTimeAggregations);
aggregationToIndexMap.put(aggregationDescriptor, deviceViewInputIndexes.get(index));
index++;
}
@@ -391,6 +400,7 @@ public class LogicalPlanBuilder {
constructSourceNodeFromAggregationDescriptors(
ascendingAggregations,
descendingAggregations,
+ countTimeAggregations,
scanOrder,
timeFilter,
groupByTimeParameter);
@@ -427,7 +437,8 @@ public class LogicalPlanBuilder {
Ordering scanOrder,
boolean needCheckAscending,
Map<PartialPath, List<AggregationDescriptor>> ascendingAggregations,
- Map<PartialPath, List<AggregationDescriptor>> descendingAggregations) {
+ Map<PartialPath, List<AggregationDescriptor>> descendingAggregations,
+ Map<PartialPath, List<AggregationDescriptor>> countTimeAggregations) {
AggregationDescriptor aggregationDescriptor =
new AggregationDescriptor(
sourceExpression.getFunctionName(),
@@ -437,6 +448,38 @@ public class LogicalPlanBuilder {
if (curStep.isOutputPartial()) {
updateTypeProviderByPartialAggregation(aggregationDescriptor, context.getTypeProvider());
}
+
+ if (COUNT_TIME.equalsIgnoreCase(sourceExpression.getFunctionName())) {
+ Map<String, Pair<List<String>, List<IMeasurementSchema>>> map = new HashMap<>();
+ for (Expression expression : sourceExpression.getCountTimeExpressions()) {
+ TimeSeriesOperand ts = (TimeSeriesOperand) expression;
+ PartialPath path = ts.getPath();
+ Pair<List<String>, List<IMeasurementSchema>> pair =
+ map.computeIfAbsent(
+ path.getDevice(), k -> new Pair<>(new ArrayList<>(), new ArrayList<>()));
+ pair.left.add(path.getMeasurement());
+ try {
+ pair.right.add(path.getMeasurementSchema());
+ } catch (MetadataException ex) {
+ throw new RuntimeException(ex);
+ }
+ }
+
+ for (Map.Entry<String, Pair<List<String>, List<IMeasurementSchema>>> entry : map.entrySet()) {
+ String device = entry.getKey();
+ Pair<List<String>, List<IMeasurementSchema>> pair = entry.getValue();
+ AlignedPath alignedPath = null;
+ try {
+ alignedPath = new AlignedPath(device, pair.left, pair.right);
+ } catch (IllegalPathException e) {
+ throw new RuntimeException(e);
+ }
+ countTimeAggregations.put(alignedPath, Collections.singletonList(aggregationDescriptor));
+ }
+
+ return aggregationDescriptor;
+ }
+
PartialPath selectPath =
((TimeSeriesOperand) sourceExpression.getExpressions().get(0)).getPath();
if (!needCheckAscending
@@ -456,13 +499,20 @@ public class LogicalPlanBuilder {
private List<PlanNode> constructSourceNodeFromAggregationDescriptors(
Map<PartialPath, List<AggregationDescriptor>> ascendingAggregations,
Map<PartialPath, List<AggregationDescriptor>> descendingAggregations,
+ Map<PartialPath, List<AggregationDescriptor>> countTimeAggregations,
Ordering scanOrder,
Filter timeFilter,
GroupByTimeParameter groupByTimeParameter) {
+
List<PlanNode> sourceNodeList = new ArrayList<>();
boolean needCheckAscending = groupByTimeParameter == null;
- Map<PartialPath, List<AggregationDescriptor>> groupedAscendingAggregations =
- MetaUtils.groupAlignedAggregations(ascendingAggregations);
+ Map<PartialPath, List<AggregationDescriptor>> groupedAscendingAggregations = null;
+ if (!countTimeAggregations.isEmpty()) {
+ groupedAscendingAggregations = countTimeAggregations;
+ } else {
+ groupedAscendingAggregations = MetaUtils.groupAlignedAggregations(ascendingAggregations);
+ }
+
for (Map.Entry<PartialPath, List<AggregationDescriptor>> pathAggregationsEntry :
groupedAscendingAggregations.entrySet()) {
sourceNodeList.add(
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/LogicalPlanVisitor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/LogicalPlanVisitor.java
index 9f28ec21026..d5180aacf46 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/LogicalPlanVisitor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/LogicalPlanVisitor.java
@@ -18,6 +18,8 @@
*/
package org.apache.iotdb.db.queryengine.plan.planner;
+import org.apache.iotdb.commons.path.AlignedPath;
+import org.apache.iotdb.commons.path.MeasurementPath;
import org.apache.iotdb.commons.path.PartialPath;
import org.apache.iotdb.commons.schema.view.viewExpression.ViewExpression;
import org.apache.iotdb.commons.udf.builtin.BuiltinAggregationFunction;
@@ -25,6 +27,7 @@ import org.apache.iotdb.db.queryengine.common.MPPQueryContext;
import org.apache.iotdb.db.queryengine.plan.analyze.Analysis;
import org.apache.iotdb.db.queryengine.plan.analyze.ExpressionAnalyzer;
import org.apache.iotdb.db.queryengine.plan.expression.Expression;
+import org.apache.iotdb.db.queryengine.plan.expression.leaf.TimeSeriesOperand;
import org.apache.iotdb.db.queryengine.plan.expression.multi.FunctionExpression;
import org.apache.iotdb.db.queryengine.plan.expression.visitor.TransformToViewExpressionVisitor;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNode;
@@ -87,6 +90,8 @@ import org.apache.iotdb.db.queryengine.plan.statement.sys.ShowQueriesStatement;
import org.apache.iotdb.db.schemaengine.template.Template;
import org.apache.iotdb.tsfile.utils.Pair;
+import org.apache.commons.lang3.StringUtils;
+
import java.util.ArrayList;
import java.util.HashMap;
import java.util.LinkedHashMap;
@@ -94,6 +99,8 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
+import static org.apache.iotdb.db.utils.constant.SqlConstant.COUNT_TIME;
+
/**
* This visitor is used to generate a logical plan for the statement and returns the {@link
* PlanNode}.
@@ -249,7 +256,7 @@ public class LogicalPlanVisitor extends StatementVisitor<PlanNode, MPPQueryConte
analysis.hasValueFilter()
|| analysis.hasGroupByParameter()
|| needTransform(sourceTransformExpressions)
- || cannotUseStatistics(aggregationExpressions);
+ || cannotUseStatistics(aggregationExpressions, sourceTransformExpressions);
AggregationStep curStep;
if (isRawDataSource) {
planBuilder =
@@ -345,11 +352,31 @@ public class LogicalPlanVisitor extends StatementVisitor<PlanNode, MPPQueryConte
return false;
}
- private boolean cannotUseStatistics(Set<Expression> expressions) {
+ private boolean cannotUseStatistics(
+ Set<Expression> expressions, Set<Expression> sourceTransformExpressions) {
for (Expression expression : expressions) {
+
if (expression instanceof FunctionExpression) {
- if (!BuiltinAggregationFunction.canUseStatistics(
- ((FunctionExpression) expression).getFunctionName())) {
+ FunctionExpression functionExpression = (FunctionExpression) expression;
+ if (COUNT_TIME.equalsIgnoreCase(functionExpression.getFunctionName())) {
+ String alignedDeviceId = "";
+ for (Expression countTimeExpression : sourceTransformExpressions) {
+ TimeSeriesOperand ts = (TimeSeriesOperand) countTimeExpression;
+ if (!(ts.getPath() instanceof AlignedPath
+ || ((MeasurementPath) ts.getPath()).isUnderAlignedEntity())) {
+ return true;
+ }
+ if (StringUtils.isEmpty(alignedDeviceId)) {
+ alignedDeviceId = ts.getPath().getDevice();
+ } else if (!alignedDeviceId.equalsIgnoreCase(ts.getPath().getDevice())) {
+ // count_time from only one aligned device can use AlignedSeriesAggScan
+ return true;
+ }
+ }
+ return false;
+ }
+
+ if (!BuiltinAggregationFunction.canUseStatistics(functionExpression.getFunctionName())) {
return true;
}
} else {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/OperatorTreeGenerator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/OperatorTreeGenerator.java
index 48d2417f8a5..3585b301e10 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/OperatorTreeGenerator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/OperatorTreeGenerator.java
@@ -136,6 +136,7 @@ import org.apache.iotdb.db.queryengine.plan.analyze.TypeProvider;
import org.apache.iotdb.db.queryengine.plan.analyze.cache.schema.DataNodeSchemaCache;
import org.apache.iotdb.db.queryengine.plan.expression.Expression;
import org.apache.iotdb.db.queryengine.plan.expression.leaf.TimeSeriesOperand;
+import org.apache.iotdb.db.queryengine.plan.expression.leaf.TimestampOperand;
import org.apache.iotdb.db.queryengine.plan.expression.visitor.ColumnTransformerVisitor;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanVisitor;
@@ -243,6 +244,7 @@ import static org.apache.iotdb.db.queryengine.common.DataNodeEndPoints.isSameNod
import static org.apache.iotdb.db.queryengine.execution.operator.AggregationUtil.calculateMaxAggregationResultSize;
import static org.apache.iotdb.db.queryengine.execution.operator.AggregationUtil.calculateMaxAggregationResultSizeForLastQuery;
import static org.apache.iotdb.db.queryengine.execution.operator.AggregationUtil.initTimeRangeIterator;
+import static org.apache.iotdb.db.queryengine.plan.expression.leaf.TimestampOperand.TIMESTAMP_EXPRESSION_STRING;
import static org.apache.iotdb.db.queryengine.plan.planner.plan.parameter.SeriesScanOptions.updateFilterUsingTTL;
/** This Visitor is responsible for transferring PlanNode Tree to Operator Tree. */
@@ -430,26 +432,42 @@ public class OperatorTreeGenerator extends PlanVisitor<Operator, LocalExecutionP
checkArgument(
descriptor.getInputExpressions().size() == 1,
"descriptor's input expression size is not 1");
- checkArgument(
- descriptor.getInputExpressions().get(0) instanceof TimeSeriesOperand,
- "descriptor's input expression is not TimeSeriesOperand");
- String inputSeries =
- ((TimeSeriesOperand) (descriptor.getInputExpressions().get(0)))
- .getPath()
- .getMeasurement();
- int seriesIndex = seriesPath.getMeasurementList().indexOf(inputSeries);
- TSDataType seriesDataType =
- seriesPath.getMeasurementSchema().getSubMeasurementsTSDataTypeList().get(seriesIndex);
- aggregators.add(
- new Aggregator(
- AccumulatorFactory.createAccumulator(
- descriptor.getAggregationType(),
- seriesDataType,
- descriptor.getInputExpressions(),
- descriptor.getInputAttributes(),
- ascending),
- descriptor.getStep(),
- Collections.singletonList(new InputLocation[] {new InputLocation(0, seriesIndex)})));
+ Expression expression = descriptor.getInputExpressions().get(0);
+ if (expression instanceof TimeSeriesOperand) {
+ String inputSeries =
+ ((TimeSeriesOperand) (descriptor.getInputExpressions().get(0)))
+ .getPath()
+ .getMeasurement();
+ int seriesIndex = seriesPath.getMeasurementList().indexOf(inputSeries);
+ TSDataType seriesDataType =
+ seriesPath.getMeasurementSchema().getSubMeasurementsTSDataTypeList().get(seriesIndex);
+ aggregators.add(
+ new Aggregator(
+ AccumulatorFactory.createAccumulator(
+ descriptor.getAggregationType(),
+ seriesDataType,
+ descriptor.getInputExpressions(),
+ descriptor.getInputAttributes(),
+ ascending),
+ descriptor.getStep(),
+ Collections.singletonList(
+ new InputLocation[] {new InputLocation(0, seriesIndex)})));
+ } else if (expression instanceof TimestampOperand) {
+ aggregators.add(
+ new Aggregator(
+ AccumulatorFactory.createAccumulator(
+ descriptor.getAggregationType(),
+ TSDataType.INT64,
+ descriptor.getInputExpressions(),
+ descriptor.getInputAttributes(),
+ ascending),
+ descriptor.getStep(),
+ Collections.singletonList(new InputLocation[] {new InputLocation(0, -1)})));
+ } else {
+ throw new IllegalArgumentException(
+ "descriptor's input expression must be TimeSeriesOperand/TimestampOperand, current is "
+ + expression);
+ }
}
GroupByTimeParameter groupByTimeParameter = node.getGroupByTimeParameter();
@@ -1015,7 +1033,7 @@ public class OperatorTreeGenerator extends PlanVisitor<Operator, LocalExecutionP
ImmutableMap.of(),
ImmutableList.of(),
inputDataTypes,
- inputLocations.size());
+ inputLocations.size() - 1);
for (Expression expression : projectExpressions) {
projectOutputTransformerList.add(
@@ -1141,7 +1159,7 @@ public class OperatorTreeGenerator extends PlanVisitor<Operator, LocalExecutionP
filterExpressionColumnTransformerMap,
commonTransformerList,
filterOutputDataTypes,
- inputLocations.size());
+ inputLocations.size() - 1);
for (Expression expression : projectExpressions) {
projectOutputTransformerList.add(
@@ -1653,7 +1671,6 @@ public class OperatorTreeGenerator extends PlanVisitor<Operator, LocalExecutionP
targetPathToDataTypeMap,
intoPathDescriptor.getTargetDeviceToAlignedMap(),
sourceTargetPathPairWithViewList,
- sourceColumnToInputLocationMap,
FragmentInstanceManager.getInstance().getIntoOperationExecutor(),
statementSizePerLine);
}
@@ -1830,8 +1847,9 @@ public class OperatorTreeGenerator extends PlanVisitor<Operator, LocalExecutionP
private List<OutputColumn> generateOutputColumnsFromChildren(MultiChildProcessNode node) {
// TODO we should also sort the InputLocation for each column if they are not overlapped
- return makeLayout(node).values().stream()
- .map(inputLocations -> new OutputColumn(inputLocations, inputLocations.size() > 1))
+ return makeLayout(node).entrySet().stream()
+ .filter(entry -> !TIMESTAMP_EXPRESSION_STRING.equals(entry.getKey()))
+ .map(entry -> new OutputColumn(entry.getValue(), entry.getValue().size() > 1))
.collect(Collectors.toList());
}
@@ -2376,6 +2394,9 @@ public class OperatorTreeGenerator extends PlanVisitor<Operator, LocalExecutionP
Map<String, List<InputLocation>> outputMappings = new LinkedHashMap<>();
int tsBlockIndex = 0;
for (PlanNode childNode : node.getChildren()) {
+ outputMappings
+ .computeIfAbsent(TIMESTAMP_EXPRESSION_STRING, key -> new ArrayList<>())
+ .add(new InputLocation(tsBlockIndex, -1));
int valueColumnIndex = 0;
for (String columnName : childNode.getOutputColumnNames()) {
outputMappings
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/crud/QueryStatement.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/crud/QueryStatement.java
index ae363342256..06458367fd4 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/crud/QueryStatement.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/crud/QueryStatement.java
@@ -26,6 +26,7 @@ import org.apache.iotdb.db.queryengine.plan.analyze.ExpressionAnalyzer;
import org.apache.iotdb.db.queryengine.plan.expression.Expression;
import org.apache.iotdb.db.queryengine.plan.expression.leaf.TimeSeriesOperand;
import org.apache.iotdb.db.queryengine.plan.expression.multi.FunctionExpression;
+import org.apache.iotdb.db.queryengine.plan.expression.visitor.CountTimeAggregationAmountVisitor;
import org.apache.iotdb.db.queryengine.plan.statement.Statement;
import org.apache.iotdb.db.queryengine.plan.statement.StatementType;
import org.apache.iotdb.db.queryengine.plan.statement.StatementVisitor;
@@ -51,6 +52,8 @@ import java.util.HashSet;
import java.util.List;
import java.util.Set;
+import static org.apache.iotdb.db.utils.constant.SqlConstant.COUNT_TIME;
+
/**
* Base class of SELECT statement.
*
@@ -114,6 +117,8 @@ public class QueryStatement extends Statement {
private boolean useWildcard = true;
+ private boolean isCountTimeAggregation = false;
+
// used for limit and offset push down optimizer, if we select all columns from aligned device, we
// can use statistics to skip
private boolean lastLevelUseWildcard = false;
@@ -489,6 +494,14 @@ public class QueryStatement extends Statement {
return useWildcard;
}
+ public void setCountTimeAggregation(boolean countTimeAggregation) {
+ this.isCountTimeAggregation = countTimeAggregation;
+ }
+
+ public boolean isCountTimeAggregation() {
+ return this.isCountTimeAggregation;
+ }
+
public boolean isLastLevelUseWildcard() {
return lastLevelUseWildcard;
}
@@ -500,6 +513,22 @@ public class QueryStatement extends Statement {
public static final String RAW_AGGREGATION_HYBRID_QUERY_ERROR_MSG =
"Raw data and aggregation hybrid query is not supported.";
+ public static final String COUNT_TIME_NOT_SUPPORT_GROUP_BY_LEVEL =
+ "Count_time aggregation function using with group by level is not supported.";
+
+ public static final String COUNT_TIME_NOT_SUPPORT_GROUP_BY_TAG =
+ "Count_time aggregation function using with group by tag is not supported.";
+
+ public static final String COUNT_TIME_NOT_SUPPORT_USED_WITH_OTHER_OPERATION =
+ "Count_time aggregation function used with arithmetic operation "
+ + "or other aggregation is not supported.";
+
+ public static final String COUNT_TIME_CAN_ONLY_EXIST_ONE_IN_SELECT =
+ "Count_time aggregation function can only exist one in select clause.";
+
+ public static final String COUNT_TIME_NOT_SUPPORT_USE_WITH_HAVING =
+ "Count_time aggregation function can not be used with having clause.";
+
@SuppressWarnings({"squid:S3776", "squid:S6541"}) // Suppress high Cognitive Complexity warning
public void semanticCheck() {
if (isAggregationQuery()) {
@@ -517,16 +546,21 @@ public class QueryStatement extends Statement {
if (resultColumn.getColumnType() != ResultColumn.ColumnType.AGGREGATION) {
throw new SemanticException(RAW_AGGREGATION_HYBRID_QUERY_ERROR_MSG);
}
+
+ String expressionString = resultColumn.getExpression().getExpressionString();
+ if (expressionString.toLowerCase().contains(COUNT_TIME)) {
+ checkCountTimeValidationInSelect(resultColumn.getExpression(), outputColumn);
+ }
outputColumn.add(
- resultColumn.getAlias() != null
- ? resultColumn.getAlias()
- : resultColumn.getExpression().getExpressionString());
+ resultColumn.getAlias() != null ? resultColumn.getAlias() : expressionString);
}
+
for (Expression expression : getExpressionSortItemList()) {
if (!hasAggregationFunction(expression)) {
throw new SemanticException(RAW_AGGREGATION_HYBRID_QUERY_ERROR_MSG);
}
}
+
if (isGroupByTag()) {
if (hasHaving()) {
throw new SemanticException("Having clause is not supported yet in GROUP BY TAGS query");
@@ -588,6 +622,10 @@ public class QueryStatement extends Statement {
throw new SemanticException(
"Expression of HAVING clause can not be used in NonAggregationQuery");
}
+ if (havingExpression.toString().toLowerCase().contains(COUNT_TIME)
+ && (!new CountTimeAggregationAmountVisitor().process(havingExpression, null).isEmpty())) {
+ throw new SemanticException(COUNT_TIME_NOT_SUPPORT_USE_WITH_HAVING);
+ }
try {
if (isGroupByLevel()) {
// check path in SELECT and HAVING only have one node
@@ -727,4 +765,37 @@ public class QueryStatement extends Statement {
public <R, C> R accept(StatementVisitor<R, C> visitor, C context) {
return visitor.visitQuery(this, context);
}
+
+ private void checkCountTimeValidationInSelect(Expression expression, Set<String> outputColumn) {
+ int countTimeAggSize = new CountTimeAggregationAmountVisitor().process(expression, null).size();
+
+ if (countTimeAggSize > 1) {
+ // e.g. select count_time(*) + count_time(*) from root.**
+ throw new SemanticException(COUNT_TIME_CAN_ONLY_EXIST_ONE_IN_SELECT);
+ } else if (countTimeAggSize == 1) {
+ // e.g. select count_time(*) / 2; select sum(*) / count_time(*)
+ if (!(expression instanceof FunctionExpression)) {
+ throw new SemanticException(COUNT_TIME_NOT_SUPPORT_USED_WITH_OTHER_OPERATION);
+ }
+
+ // e.g. select count(*), count(*) from root.db.**
+ if (!outputColumn.isEmpty()) {
+ throw new SemanticException(COUNT_TIME_CAN_ONLY_EXIST_ONE_IN_SELECT);
+ }
+
+ if (isGroupByTag()) {
+ throw new SemanticException(COUNT_TIME_NOT_SUPPORT_GROUP_BY_TAG);
+ }
+
+ if (isGroupByLevel()) {
+ throw new SemanticException(COUNT_TIME_NOT_SUPPORT_GROUP_BY_LEVEL);
+ }
+
+ setCountTimeAggregation(true);
+
+ if (hasHaving()) {
+ throw new SemanticException(COUNT_TIME_NOT_SUPPORT_USE_WITH_HAVING);
+ }
+ }
+ }
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/utils/ResourceByPathUtils.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/utils/ResourceByPathUtils.java
index ff0eed48431..f17c3f90da5 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/utils/ResourceByPathUtils.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/utils/ResourceByPathUtils.java
@@ -56,6 +56,7 @@ import org.apache.iotdb.tsfile.write.writer.RestorableTsFileIOWriter;
import java.io.IOException;
import java.io.Serializable;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.List;
import java.util.Map;
@@ -116,8 +117,7 @@ class AlignedResourceByPathUtils extends ResourceByPathUtils {
*/
@Override
public AlignedTimeSeriesMetadata generateTimeSeriesMetadata(
- List<ReadOnlyMemChunk> readOnlyMemChunk, List<IChunkMetadata> chunkMetadataList)
- throws IOException {
+ List<ReadOnlyMemChunk> readOnlyMemChunk, List<IChunkMetadata> chunkMetadataList) {
TimeseriesMetadata timeTimeSeriesMetadata = new TimeseriesMetadata();
timeTimeSeriesMetadata.setOffsetOfChunkMetaDataList(-1);
timeTimeSeriesMetadata.setDataSizeOfChunkMetaDataList(-1);
@@ -278,17 +278,23 @@ class AlignedResourceByPathUtils extends ResourceByPathUtils {
}
for (int i = 0; i < timeChunkMetadataList.size(); i++) {
- List<IChunkMetadata> valueChunkMetadata = new ArrayList<>();
- // if all the sub sensors doesn't exist, it will be false
- boolean exits = false;
- for (List<ChunkMetadata> chunkMetadata : valueChunkMetadataList) {
- boolean currentExist = i < chunkMetadata.size();
- exits = (exits || currentExist);
- valueChunkMetadata.add(currentExist ? chunkMetadata.get(i) : null);
- }
- if (exits) {
+ // only need time column
+ if (partialPath.getMeasurementList().isEmpty()) {
chunkMetadataList.add(
- new AlignedChunkMetadata(timeChunkMetadataList.get(i), valueChunkMetadata));
+ new AlignedChunkMetadata(timeChunkMetadataList.get(i), Collections.emptyList()));
+ } else {
+ List<IChunkMetadata> valueChunkMetadata = new ArrayList<>();
+ // if all the sub sensors doesn't exist, it will be false
+ boolean exits = false;
+ for (List<ChunkMetadata> chunkMetadata : valueChunkMetadataList) {
+ boolean currentExist = i < chunkMetadata.size();
+ exits = (exits || currentExist);
+ valueChunkMetadata.add(currentExist ? chunkMetadata.get(i) : null);
+ }
+ if (exits) {
+ chunkMetadataList.add(
+ new AlignedChunkMetadata(timeChunkMetadataList.get(i), valueChunkMetadata));
+ }
}
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/view/ViewPaths.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/view/ViewPaths.java
index 9bee906eaa5..3aead422810 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/view/ViewPaths.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/view/ViewPaths.java
@@ -109,7 +109,7 @@ public class ViewPaths {
for (Expression expression : expressionsList) {
List<Expression> aggList = collectAggExpVisitor.process(expression, null);
- if (aggList.size() > 0) {
+ if (!aggList.isEmpty()) {
return new Pair<>(false, new ViewContainsAggregationException(aggList.get(0).toString()));
}
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/TsFileResource.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/TsFileResource.java
index 9ccf2f4aed3..d31067b5a47 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/TsFileResource.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/TsFileResource.java
@@ -205,23 +205,6 @@ public class TsFileResource {
this.tierLevel = new AtomicInteger(0);
}
- /** unsealed TsFile, for read */
- public TsFileResource(
- PartialPath path,
- List<ReadOnlyMemChunk> readOnlyMemChunk,
- List<IChunkMetadata> chunkMetadataList,
- TsFileResource originTsFileResource)
- throws IOException {
- this.file = originTsFileResource.file;
- this.timeIndex = originTsFileResource.timeIndex;
- this.pathToReadOnlyMemChunkMap.put(path, readOnlyMemChunk);
- this.pathToChunkMetadataListMap.put(path, chunkMetadataList);
- this.originTsFileResource = originTsFileResource;
- this.version = originTsFileResource.version;
- this.isSeq = originTsFileResource.isSeq;
- this.tierLevel = originTsFileResource.tierLevel;
- }
-
/** unsealed TsFile, for read */
public TsFileResource(
Map<PartialPath, List<ReadOnlyMemChunk>> pathToReadOnlyMemChunkMap,
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/FileLoaderUtils.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/FileLoaderUtils.java
index 55d9f04e30e..2d6d53809da 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/FileLoaderUtils.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/FileLoaderUtils.java
@@ -48,6 +48,7 @@ import org.apache.iotdb.tsfile.write.writer.TsFileIOWriter;
import java.io.IOException;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
@@ -285,26 +286,35 @@ public class FileLoaderUtils {
resource.getTimeIndexType() != 1,
isDebug);
if (timeColumn != null) {
- List<TimeseriesMetadata> valueTimeSeriesMetadataList =
- new ArrayList<>(valueMeasurementList.size());
- // if all the queried aligned sensors does not exist, we will return null
- boolean exist = false;
- for (String valueMeasurement : valueMeasurementList) {
- TimeseriesMetadata valueColumn =
- cache.get(
- new TimeSeriesMetadataCacheKey(filePath, deviceId, valueMeasurement),
- allSensors,
- resource.getTimeIndexType() != 1,
- isDebug);
- exist = (exist || (valueColumn != null));
- valueTimeSeriesMetadataList.add(valueColumn);
- }
- if (exist) {
+ // only need time column, like count_time aggregation
+ if (valueMeasurementList.isEmpty()) {
alignedTimeSeriesMetadata =
- new AlignedTimeSeriesMetadata(timeColumn, valueTimeSeriesMetadataList);
+ new AlignedTimeSeriesMetadata(timeColumn, Collections.emptyList());
alignedTimeSeriesMetadata.setChunkMetadataLoader(
new DiskAlignedChunkMetadataLoader(
resource, alignedPath, context, filter, queryAllSensors));
+ } else {
+ List<TimeseriesMetadata> valueTimeSeriesMetadataList =
+ new ArrayList<>(valueMeasurementList.size());
+ // if all the queried aligned sensors does not exist, we will return null
+ boolean exist = false;
+ for (String valueMeasurement : valueMeasurementList) {
+ TimeseriesMetadata valueColumn =
+ cache.get(
+ new TimeSeriesMetadataCacheKey(filePath, deviceId, valueMeasurement),
+ allSensors,
+ resource.getTimeIndexType() != 1,
+ isDebug);
+ exist = (exist || (valueColumn != null));
+ valueTimeSeriesMetadataList.add(valueColumn);
+ }
+ if (exist) {
+ alignedTimeSeriesMetadata =
+ new AlignedTimeSeriesMetadata(timeColumn, valueTimeSeriesMetadataList);
+ alignedTimeSeriesMetadata.setChunkMetadataLoader(
+ new DiskAlignedChunkMetadataLoader(
+ resource, alignedPath, context, filter, queryAllSensors));
+ }
}
}
return alignedTimeSeriesMetadata;
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/SchemaUtils.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/SchemaUtils.java
index b00ec481b37..efae314ecaa 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/SchemaUtils.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/SchemaUtils.java
@@ -120,6 +120,7 @@ public class SchemaUtils {
case SqlConstant.MAX_TIME:
case SqlConstant.COUNT:
case SqlConstant.TIME_DURATION:
+ case SqlConstant.COUNT_TIME:
return TSDataType.INT64;
case SqlConstant.AVG:
case SqlConstant.SUM:
@@ -153,6 +154,7 @@ public class SchemaUtils {
case MAX_VALUE:
case EXTREME:
case COUNT:
+ case COUNT_TIME:
case AVG:
case TIME_DURATION:
return true;
@@ -185,6 +187,7 @@ public class SchemaUtils {
case MAX_VALUE:
case EXTREME:
case COUNT:
+ case COUNT_TIME:
case MIN_TIME:
case MAX_TIME:
case COUNT_IF:
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/TypeInferenceUtils.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/TypeInferenceUtils.java
index 5f7fa237b86..9d23177b20b 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/TypeInferenceUtils.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/TypeInferenceUtils.java
@@ -24,7 +24,6 @@ import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.exception.sql.SemanticException;
import org.apache.iotdb.db.queryengine.plan.analyze.ExpressionUtils;
import org.apache.iotdb.db.queryengine.plan.expression.Expression;
-import org.apache.iotdb.db.queryengine.plan.expression.binary.BinaryExpression;
import org.apache.iotdb.db.queryengine.plan.expression.binary.CompareBinaryExpression;
import org.apache.iotdb.db.queryengine.plan.expression.leaf.ConstantOperand;
import org.apache.iotdb.db.queryengine.plan.expression.leaf.TimeSeriesOperand;
@@ -133,6 +132,7 @@ public class TypeInferenceUtils {
case SqlConstant.MIN_TIME:
case SqlConstant.MAX_TIME:
case SqlConstant.COUNT:
+ case SqlConstant.COUNT_TIME:
case SqlConstant.COUNT_IF:
case SqlConstant.TIME_DURATION:
return TSDataType.INT64;
@@ -168,6 +168,7 @@ public class TypeInferenceUtils {
throw new SemanticException(
"Aggregate functions [AVG, SUM, EXTREME, MIN_VALUE, MAX_VALUE] only support numeric data types [INT32, INT64, FLOAT, DOUBLE]");
case SqlConstant.COUNT:
+ case SqlConstant.COUNT_TIME:
case SqlConstant.MIN_TIME:
case SqlConstant.MAX_TIME:
case SqlConstant.FIRST_VALUE:
@@ -205,6 +206,7 @@ public class TypeInferenceUtils {
case SqlConstant.MIN_VALUE:
case SqlConstant.MAX_VALUE:
case SqlConstant.COUNT:
+ case SqlConstant.COUNT_TIME:
case SqlConstant.MIN_TIME:
case SqlConstant.MAX_TIME:
case SqlConstant.FIRST_VALUE:
@@ -228,7 +230,7 @@ public class TypeInferenceUtils {
outputExpressionLists.add(
Collections.singletonList(
ExpressionUtils.reconstructBinaryExpression(
- (BinaryExpression) keepExpression,
+ keepExpression,
new TimeSeriesOperand(
new MeasurementPath(
((TimeSeriesOperand) leftExpression).getPath(), TSDataType.INT64)),
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/constant/SqlConstant.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/constant/SqlConstant.java
index b8422d90139..afd47cc8c51 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/constant/SqlConstant.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/constant/SqlConstant.java
@@ -57,6 +57,9 @@ public class SqlConstant {
public static final String TIME_DURATION = "time_duration";
public static final String MODE = "mode";
+ public static final String COUNT_TIME = "count_time";
+ public static final String COUNT_TIME_HEADER = "count_time(*)";
+
// names of scalar functions
public static final String DIFF = "diff";
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/aggregation/AccumulatorTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/aggregation/AccumulatorTest.java
index 57915a27f29..c5458553b00 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/aggregation/AccumulatorTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/aggregation/AccumulatorTest.java
@@ -22,6 +22,7 @@ package org.apache.iotdb.db.queryengine.execution.aggregation;
import org.apache.iotdb.common.rpc.thrift.TAggregationType;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.file.metadata.statistics.Statistics;
+import org.apache.iotdb.tsfile.file.metadata.statistics.TimeStatistics;
import org.apache.iotdb.tsfile.read.common.block.TsBlock;
import org.apache.iotdb.tsfile.read.common.block.TsBlockBuilder;
import org.apache.iotdb.tsfile.read.common.block.column.Column;
@@ -158,6 +159,49 @@ public class AccumulatorTest {
Assert.assertEquals(1, finalResult.build().getLong(0));
}
+ @Test
+ public void countTimeAccumulatorTest() {
+ Accumulator countTimeAccumulator =
+ AccumulatorFactory.createAccumulator(
+ TAggregationType.COUNT_TIME,
+ TSDataType.DOUBLE,
+ Collections.emptyList(),
+ Collections.emptyMap(),
+ true);
+ Assert.assertEquals(TSDataType.INT64, countTimeAccumulator.getIntermediateType()[0]);
+ Assert.assertEquals(TSDataType.INT64, countTimeAccumulator.getFinalType());
+ // check returning null while no data
+ ColumnBuilder[] intermediateResult = new ColumnBuilder[1];
+ intermediateResult[0] = new LongColumnBuilder(null, 1);
+ countTimeAccumulator.outputIntermediate(intermediateResult);
+ Assert.assertEquals(0, intermediateResult[0].build().getLong(0));
+ ColumnBuilder finalResult = new LongColumnBuilder(null, 1);
+ countTimeAccumulator.outputFinal(finalResult);
+ Assert.assertEquals(0, finalResult.build().getLong(0));
+
+ Column[] timeAndValueColumn = getTimeAndValueColumn(0);
+ timeAndValueColumn[1] = timeAndValueColumn[0];
+ countTimeAccumulator.addInput(timeAndValueColumn, null, rawData.getPositionCount() - 1);
+ Assert.assertFalse(countTimeAccumulator.hasFinalResult());
+ intermediateResult[0] = new LongColumnBuilder(null, 1);
+ countTimeAccumulator.outputIntermediate(intermediateResult);
+ Assert.assertEquals(100, intermediateResult[0].build().getLong(0));
+
+ // add intermediate result as input
+ countTimeAccumulator.addIntermediate(new Column[] {intermediateResult[0].build()});
+ finalResult = new LongColumnBuilder(null, 1);
+ countTimeAccumulator.outputFinal(finalResult);
+ Assert.assertEquals(200, finalResult.build().getLong(0));
+
+ countTimeAccumulator.reset();
+ TimeStatistics timeStatistics = new TimeStatistics();
+ timeStatistics.update(100L);
+ countTimeAccumulator.addStatistics(timeStatistics);
+ finalResult = new LongColumnBuilder(null, 1);
+ countTimeAccumulator.outputFinal(finalResult);
+ Assert.assertEquals(1, finalResult.build().getLong(0));
+ }
+
@Test
public void extremeAccumulatorTest() {
Accumulator extremeAccumulator =
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/OperatorMemoryTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/OperatorMemoryTest.java
index ad3065bc6ee..539d90bc84d 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/OperatorMemoryTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/OperatorMemoryTest.java
@@ -1496,7 +1496,6 @@ public class OperatorMemoryTest {
Collections.emptyMap(),
Collections.emptyMap(),
Collections.emptyList(),
- Collections.emptyMap(),
null,
statementSizePerLine);
}
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/analyze/AnalyzeFailTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/analyze/AnalyzeFailTest.java
index b287fa723c9..9fd663e3c65 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/analyze/AnalyzeFailTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/analyze/AnalyzeFailTest.java
@@ -29,12 +29,18 @@ import org.junit.Test;
import java.time.ZonedDateTime;
+import static org.apache.iotdb.db.queryengine.plan.expression.visitor.CountTimeAggregationAmountVisitor.COUNT_TIME_ONLY_SUPPORT_ONE_WILDCARD;
import static org.apache.iotdb.db.queryengine.plan.statement.component.IntoComponent.DEVICE_ALIGNMENT_INCONSISTENT_ERROR_MSG;
import static org.apache.iotdb.db.queryengine.plan.statement.component.IntoComponent.DEVICE_NUM_MISMATCH_ERROR_MSG;
import static org.apache.iotdb.db.queryengine.plan.statement.component.IntoComponent.DUPLICATE_TARGET_PATH_ERROR_MSG;
import static org.apache.iotdb.db.queryengine.plan.statement.component.IntoComponent.FORBID_PLACEHOLDER_ERROR_MSG;
import static org.apache.iotdb.db.queryengine.plan.statement.component.IntoComponent.PATH_NUM_MISMATCH_ERROR_MSG;
import static org.apache.iotdb.db.queryengine.plan.statement.component.IntoComponent.PLACEHOLDER_MISMATCH_ERROR_MSG;
+import static org.apache.iotdb.db.queryengine.plan.statement.crud.QueryStatement.COUNT_TIME_CAN_ONLY_EXIST_ONE_IN_SELECT;
+import static org.apache.iotdb.db.queryengine.plan.statement.crud.QueryStatement.COUNT_TIME_NOT_SUPPORT_GROUP_BY_LEVEL;
+import static org.apache.iotdb.db.queryengine.plan.statement.crud.QueryStatement.COUNT_TIME_NOT_SUPPORT_GROUP_BY_TAG;
+import static org.apache.iotdb.db.queryengine.plan.statement.crud.QueryStatement.COUNT_TIME_NOT_SUPPORT_USED_WITH_OTHER_OPERATION;
+import static org.apache.iotdb.db.queryengine.plan.statement.crud.QueryStatement.COUNT_TIME_NOT_SUPPORT_USE_WITH_HAVING;
import static org.junit.Assert.fail;
public class AnalyzeFailTest {
@@ -150,6 +156,72 @@ public class AnalyzeFailTest {
PLACEHOLDER_MISMATCH_ERROR_MSG);
}
+ @Test
+ public void countTimeTest() {
+ assertAnalyzeSemanticException(
+ "select count_time(*) from root.sg.* group by level=1;",
+ COUNT_TIME_NOT_SUPPORT_GROUP_BY_LEVEL);
+
+ assertAnalyzeSemanticException(
+ "select count_time(*) from root.sg.* group by tags(key);",
+ COUNT_TIME_NOT_SUPPORT_GROUP_BY_TAG);
+
+ assertAnalyzeSemanticException(
+ "select count_time(s1) from root.sg.*;", COUNT_TIME_ONLY_SUPPORT_ONE_WILDCARD);
+
+ assertAnalyzeSemanticException(
+ "select count_time(s1) from root.sg.d1;", COUNT_TIME_ONLY_SUPPORT_ONE_WILDCARD);
+
+ assertAnalyzeSemanticException(
+ "select count_time(d1.s1) from root.sg.*;", COUNT_TIME_ONLY_SUPPORT_ONE_WILDCARD);
+
+ assertAnalyzeSemanticException(
+ "select count_time(d1.*) from root.sg.*;", COUNT_TIME_ONLY_SUPPORT_ONE_WILDCARD);
+
+ assertAnalyzeSemanticException(
+ "select count_time(* + *) from root.sg.*;", COUNT_TIME_ONLY_SUPPORT_ONE_WILDCARD);
+
+ assertAnalyzeSemanticException(
+ "select count_time(*) / 2 from root.sg.*;",
+ COUNT_TIME_NOT_SUPPORT_USED_WITH_OTHER_OPERATION);
+
+ assertAnalyzeSemanticException(
+ "select sum(s1) / count_time(*) from root.sg.*;",
+ COUNT_TIME_NOT_SUPPORT_USED_WITH_OTHER_OPERATION);
+
+ assertAnalyzeSemanticException(
+ "select count_time(*),count_time(*) from root.sg.**;",
+ COUNT_TIME_CAN_ONLY_EXIST_ONE_IN_SELECT);
+
+ assertAnalyzeSemanticException(
+ "select count_time(*),count_time(*) from root.sg.d1,root.sg.d2;",
+ COUNT_TIME_CAN_ONLY_EXIST_ONE_IN_SELECT);
+
+ assertAnalyzeSemanticException(
+ "select COUNT_TIME(*),COUNT_TIME(*) from root.sg.d1,root.sg.d2;",
+ COUNT_TIME_CAN_ONLY_EXIST_ONE_IN_SELECT);
+
+ assertAnalyzeSemanticException(
+ "select COUNT_TIME(*),count_time(*) from root.sg.d1,root.sg.d2;",
+ COUNT_TIME_CAN_ONLY_EXIST_ONE_IN_SELECT);
+
+ assertAnalyzeSemanticException(
+ "select COUNT_TIME(*),COUNT_time(*) from root.sg.d1,root.sg.d2;",
+ COUNT_TIME_CAN_ONLY_EXIST_ONE_IN_SELECT);
+
+ assertAnalyzeSemanticException(
+ "select count_time(*) from root.sg.* having count(*) > 1;",
+ COUNT_TIME_NOT_SUPPORT_USE_WITH_HAVING);
+
+ assertAnalyzeSemanticException(
+ "select count_time(*) from root.sg.* having count_time(s1) > 1;",
+ COUNT_TIME_NOT_SUPPORT_USE_WITH_HAVING);
+
+ assertAnalyzeSemanticException(
+ "select count_time(*) from root.sg.* having count_time(*) > 1;",
+ COUNT_TIME_NOT_SUPPORT_USE_WITH_HAVING);
+ }
+
private void assertAnalyzeSemanticException(String sql, String message) {
try {
Analyzer analyzer =
diff --git a/iotdb-core/datanode/src/test/resources/datanode1conf/iotdb-common.properties b/iotdb-core/datanode/src/test/resources/datanode1conf/iotdb-common.properties
index ba222515546..b89ef13778b 100644
--- a/iotdb-core/datanode/src/test/resources/datanode1conf/iotdb-common.properties
+++ b/iotdb-core/datanode/src/test/resources/datanode1conf/iotdb-common.properties
@@ -20,4 +20,4 @@
timestamp_precision=ms
udf_lib_dir=target/datanode1/ext/udf
trigger_lib_dir=target/datanode1/ext/trigger
-pipe_lib_dir=target/datanode1/ext/pipe
+pipe_lib_dir=target/datanode1/ext/pipe
\ No newline at end of file
diff --git a/iotdb-core/datanode/src/test/resources/datanode2conf/iotdb-common.properties b/iotdb-core/datanode/src/test/resources/datanode2conf/iotdb-common.properties
index d75d1b52021..9cf060d61fd 100644
--- a/iotdb-core/datanode/src/test/resources/datanode2conf/iotdb-common.properties
+++ b/iotdb-core/datanode/src/test/resources/datanode2conf/iotdb-common.properties
@@ -16,8 +16,7 @@
# specific language governing permissions and limitations
# under the License.
#
-
timestamp_precision=ms
udf_lib_dir=target/datanode2/ext/udf
trigger_lib_dir=target/datanode2/ext/trigger
-pipe_lib_dir=target/datanode2/ext/pipe
+pipe_lib_dir=target/datanode2/ext/pipe
\ No newline at end of file
diff --git a/iotdb-core/datanode/src/test/resources/datanode3conf/iotdb-common.properties b/iotdb-core/datanode/src/test/resources/datanode3conf/iotdb-common.properties
index ebd7d4fa59f..83dbc1b051e 100644
--- a/iotdb-core/datanode/src/test/resources/datanode3conf/iotdb-common.properties
+++ b/iotdb-core/datanode/src/test/resources/datanode3conf/iotdb-common.properties
@@ -20,4 +20,4 @@
timestamp_precision=ms
udf_lib_dir=target/datanode3/ext/udf
trigger_lib_dir=target/datanode3/ext/trigger
-pipe_lib_dir=target/datanode3/ext/pipe
+pipe_lib_dir=target/datanode3/ext/pipe
\ No newline at end of file
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/BuiltinAggregationFunction.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/BuiltinAggregationFunction.java
index 43181be7aba..1afda7f2a53 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/BuiltinAggregationFunction.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/BuiltinAggregationFunction.java
@@ -37,7 +37,8 @@ public enum BuiltinAggregationFunction {
SUM("sum"),
COUNT_IF("count_if"),
TIME_DURATION("time_duration"),
- MODE("mode");
+ MODE("mode"),
+ COUNT_TIME("count_time");
private final String functionName;
@@ -77,6 +78,7 @@ public enum BuiltinAggregationFunction {
return true;
case "count_if":
case "mode":
+ case "count_time":
return false;
default:
throw new IllegalArgumentException("Invalid Aggregation function: " + name);
@@ -103,6 +105,7 @@ public enum BuiltinAggregationFunction {
case "mode":
return true;
case "count_if":
+ case "count_time":
return false;
default:
throw new IllegalArgumentException("Invalid Aggregation function: " + name);
diff --git a/iotdb-core/tsfile/src/main/java/org/apache/iotdb/tsfile/file/metadata/AlignedTimeSeriesMetadata.java b/iotdb-core/tsfile/src/main/java/org/apache/iotdb/tsfile/file/metadata/AlignedTimeSeriesMetadata.java
index e1d32300654..79d20588418 100644
--- a/iotdb-core/tsfile/src/main/java/org/apache/iotdb/tsfile/file/metadata/AlignedTimeSeriesMetadata.java
+++ b/iotdb-core/tsfile/src/main/java/org/apache/iotdb/tsfile/file/metadata/AlignedTimeSeriesMetadata.java
@@ -23,6 +23,7 @@ import org.apache.iotdb.tsfile.file.metadata.statistics.Statistics;
import org.apache.iotdb.tsfile.read.controller.IChunkMetadataLoader;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.List;
public class AlignedTimeSeriesMetadata implements ITimeSeriesMetadata {
@@ -135,20 +136,25 @@ public class AlignedTimeSeriesMetadata implements ITimeSeriesMetadata {
List<IChunkMetadata> timeChunkMetadata, List<List<IChunkMetadata>> valueChunkMetadataList) {
List<AlignedChunkMetadata> res = new ArrayList<>();
for (int i = 0; i < timeChunkMetadata.size(); i++) {
- List<IChunkMetadata> chunkMetadataList = new ArrayList<>();
- // only at least one sensor exits, we add the AlignedChunkMetadata to the list
- boolean exits = false;
- for (List<IChunkMetadata> chunkMetadata : valueChunkMetadataList) {
- IChunkMetadata v =
- chunkMetadata == null
- || chunkMetadata.get(i).getStatistics().getCount() == 0 // empty chunk
- ? null
- : chunkMetadata.get(i);
- exits = (exits || v != null);
- chunkMetadataList.add(v);
- }
- if (exits) {
- res.add(new AlignedChunkMetadata(timeChunkMetadata.get(i), chunkMetadataList));
+ // only need time column
+ if (valueTimeseriesMetadataList.isEmpty()) {
+ res.add(new AlignedChunkMetadata(timeChunkMetadata.get(i), Collections.emptyList()));
+ } else {
+ List<IChunkMetadata> chunkMetadataList = new ArrayList<>();
+ // only at least one sensor exits, we add the AlignedChunkMetadata to the list
+ boolean exits = false;
+ for (List<IChunkMetadata> chunkMetadata : valueChunkMetadataList) {
+ IChunkMetadata v =
+ chunkMetadata == null
+ || chunkMetadata.get(i).getStatistics().getCount() == 0 // empty chunk
+ ? null
+ : chunkMetadata.get(i);
+ exits = (exits || v != null);
+ chunkMetadataList.add(v);
+ }
+ if (exits) {
+ res.add(new AlignedChunkMetadata(timeChunkMetadata.get(i), chunkMetadataList));
+ }
}
}
return res;
diff --git a/iotdb-core/tsfile/src/main/java/org/apache/iotdb/tsfile/read/reader/chunk/AlignedChunkReader.java b/iotdb-core/tsfile/src/main/java/org/apache/iotdb/tsfile/read/reader/chunk/AlignedChunkReader.java
index d2751df96c9..b7c79f7a1a4 100644
--- a/iotdb-core/tsfile/src/main/java/org/apache/iotdb/tsfile/read/reader/chunk/AlignedChunkReader.java
+++ b/iotdb-core/tsfile/src/main/java/org/apache/iotdb/tsfile/read/reader/chunk/AlignedChunkReader.java
@@ -148,7 +148,7 @@ public class AlignedChunkReader implements IChunkReader {
PageHeader timePageHeader;
List<PageHeader> valuePageHeaderList = new ArrayList<>();
- boolean exits = false;
+ boolean exits = valueChunkDataBufferList.isEmpty();
// this chunk has only one page
if ((timeChunkHeader.getChunkType() & 0x3F) == MetaMarker.ONLY_ONE_PAGE_CHUNK_HEADER) {
timePageHeader = PageHeader.deserializeFrom(timeChunkDataBuffer, timeChunkStatistics);
@@ -225,7 +225,7 @@ public class AlignedChunkReader implements IChunkReader {
List<ByteBuffer> valuePageDataList = new ArrayList<>();
List<TSDataType> valueDataTypeList = new ArrayList<>();
List<Decoder> valueDecoderList = new ArrayList<>();
- boolean exist = false;
+ boolean exist = valuePageHeader.isEmpty();
for (int i = 0; i < valuePageHeader.size(); i++) {
if (valuePageHeader.get(i) == null
|| valuePageHeader.get(i).getUncompressedSize() == 0) { // Empty Page
diff --git a/iotdb-core/tsfile/src/main/java/org/apache/iotdb/tsfile/read/reader/page/AlignedPageReader.java b/iotdb-core/tsfile/src/main/java/org/apache/iotdb/tsfile/read/reader/page/AlignedPageReader.java
index 8aaa3484f83..608a3e8fcb0 100644
--- a/iotdb-core/tsfile/src/main/java/org/apache/iotdb/tsfile/read/reader/page/AlignedPageReader.java
+++ b/iotdb-core/tsfile/src/main/java/org/apache/iotdb/tsfile/read/reader/page/AlignedPageReader.java
@@ -180,40 +180,43 @@ public class AlignedPageReader implements IPageReader, IAlignedPageReader {
}
}
- // using bitMap in valuePageReaders to indicate whether columns of current row are all null.
- byte[] bitmask = new byte[(timeBatch.length - 1) / 8 + 1];
- Arrays.fill(bitmask, (byte) 0x00);
- boolean[][] isDeleted = new boolean[valueCount][timeBatch.length];
- for (int columnIndex = 0; columnIndex < valueCount; columnIndex++) {
- ValuePageReader pageReader = valuePageReaderList.get(columnIndex);
- if (pageReader != null) {
- byte[] bitmap = pageReader.getBitmap();
- pageReader.fillIsDeleted(timeBatch, isDeleted[columnIndex]);
+ boolean[][] isDeleted = null;
+ if (valueCount != 0) {
+ // using bitMap in valuePageReaders to indicate whether columns of current row are all null.
+ byte[] bitmask = new byte[(timeBatch.length - 1) / 8 + 1];
+ Arrays.fill(bitmask, (byte) 0x00);
+ isDeleted = new boolean[valueCount][timeBatch.length];
+ for (int columnIndex = 0; columnIndex < valueCount; columnIndex++) {
+ ValuePageReader pageReader = valuePageReaderList.get(columnIndex);
+ if (pageReader != null) {
+ byte[] bitmap = pageReader.getBitmap();
+ pageReader.fillIsDeleted(timeBatch, isDeleted[columnIndex]);
- for (int i = 0, n = isDeleted[columnIndex].length; i < n; i++) {
- if (isDeleted[columnIndex][i]) {
- int shift = i % 8;
- bitmap[i / 8] = (byte) (bitmap[i / 8] & (~(MASK >>> shift)));
+ for (int i = 0, n = isDeleted[columnIndex].length; i < n; i++) {
+ if (isDeleted[columnIndex][i]) {
+ int shift = i % 8;
+ bitmap[i / 8] = (byte) (bitmap[i / 8] & (~(MASK >>> shift)));
+ }
+ }
+ for (int i = 0, n = bitmask.length; i < n; i++) {
+ bitmask[i] = (byte) (bitmap[i] | bitmask[i]);
}
- }
- for (int i = 0, n = bitmask.length; i < n; i++) {
- bitmask[i] = (byte) (bitmap[i] | bitmask[i]);
}
}
- }
- for (int i = 0, n = bitmask.length; i < n; i++) {
- if (bitmask[i] == (byte) 0xFF) {
- // 8 rows are not all null, do nothing
- } else if (bitmask[i] == (byte) 0x00) {
- for (int j = 0; j < 8 && (i * 8 + j < keepCurrentRow.length); j++) {
- keepCurrentRow[i * 8 + j] = false;
- }
- } else {
- for (int j = 0; j < 8 && (i * 8 + j < keepCurrentRow.length); j++) {
- if (((bitmask[i] & 0xFF) & (MASK >>> j)) == 0) {
+ for (int i = 0, n = bitmask.length; i < n; i++) {
+ if (bitmask[i] == (byte) 0xFF) {
+ // 8 rows are not all null, do nothing
+ } else if (bitmask[i] == (byte) 0x00) {
+ for (int j = 0; j < 8 && (i * 8 + j < keepCurrentRow.length); j++) {
keepCurrentRow[i * 8 + j] = false;
}
+ } else {
+ for (int j = 0; j < 8 && (i * 8 + j < keepCurrentRow.length); j++) {
+ if (((bitmask[i] & 0xFF) & (MASK >>> j)) == 0) {
+ keepCurrentRow[i * 8 + j] = false;
+ }
+ }
}
}
}
@@ -221,21 +224,18 @@ public class AlignedPageReader implements IPageReader, IAlignedPageReader {
// construct time column
int readEndIndex = timeBatch.length;
for (int i = 0; i < timeBatch.length; i++) {
- if (!keepCurrentRow[i]) {
- continue;
- }
- if (paginationController.hasCurOffset()) {
- paginationController.consumeOffset();
- keepCurrentRow[i] = false;
- continue;
- }
- if (paginationController.hasCurLimit()) {
- builder.getTimeColumnBuilder().writeLong(timeBatch[i]);
- builder.declarePosition();
- paginationController.consumeLimit();
- } else {
- readEndIndex = i;
- break;
+ if (keepCurrentRow[i]) {
+ if (paginationController.hasCurOffset()) {
+ paginationController.consumeOffset();
+ keepCurrentRow[i] = false;
+ } else if (paginationController.hasCurLimit()) {
+ builder.getTimeColumnBuilder().writeLong(timeBatch[i]);
+ builder.declarePosition();
+ paginationController.consumeLimit();
+ } else {
+ readEndIndex = i;
+ break;
+ }
}
}
diff --git a/iotdb-core/tsfile/src/main/java/org/apache/iotdb/tsfile/write/chunk/AlignedChunkWriterImpl.java b/iotdb-core/tsfile/src/main/java/org/apache/iotdb/tsfile/write/chunk/AlignedChunkWriterImpl.java
index f4b5794dc20..c71ffc3d04a 100644
--- a/iotdb-core/tsfile/src/main/java/org/apache/iotdb/tsfile/write/chunk/AlignedChunkWriterImpl.java
+++ b/iotdb-core/tsfile/src/main/java/org/apache/iotdb/tsfile/write/chunk/AlignedChunkWriterImpl.java
@@ -49,7 +49,7 @@ public class AlignedChunkWriterImpl implements IChunkWriter {
// Used for batch writing
private long remainingPointsNumber;
- /** @param schema schema of this measurement */
+ // TestOnly
public AlignedChunkWriterImpl(VectorMeasurementSchema schema) {
timeChunkWriter =
new TimeChunkWriter(
diff --git a/iotdb-protocol/thrift-commons/src/main/thrift/common.thrift b/iotdb-protocol/thrift-commons/src/main/thrift/common.thrift
index 2c07f7493c0..d602c505319 100644
--- a/iotdb-protocol/thrift-commons/src/main/thrift/common.thrift
+++ b/iotdb-protocol/thrift-commons/src/main/thrift/common.thrift
@@ -178,7 +178,8 @@ enum TAggregationType {
EXTREME,
COUNT_IF,
TIME_DURATION,
- MODE
+ MODE,
+ COUNT_TIME
}
// for MLNode