This is an automated email from the ASF dual-hosted git repository.
jackietien 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 49e5b2f [IOTDB-1951] Support aggregation query without value filter
in new vector (#4441)
49e5b2f is described below
commit 49e5b2ffeee2b36589a41eafc431da1e70d784f1
Author: Chen YZ <[email protected]>
AuthorDate: Tue Nov 23 13:10:54 2021 +0800
[IOTDB-1951] Support aggregation query without value filter in new vector
(#4441)
---
.../apache/iotdb/db/metadata/path/AlignedPath.java | 6 +-
.../db/query/aggregation/impl/CountAggrResult.java | 1 +
.../db/query/executor/AggregationExecutor.java | 20 +-
.../reader/series/VectorSeriesAggregateReader.java | 24 ++
.../IoTDBAggregationWithoutValueFilter2IT.java | 63 +++
.../IoTDBAggregationWithoutValueFilterIT.java | 444 ++++++++++++++++++++
...gregationWithoutValueFilterWithDeletion2IT.java | 83 ++++
...ggregationWithoutValueFilterWithDeletionIT.java | 447 +++++++++++++++++++++
.../apache/iotdb/tsfile/read/common/BatchData.java | 16 +
9 files changed, 1095 insertions(+), 9 deletions(-)
diff --git
a/server/src/main/java/org/apache/iotdb/db/metadata/path/AlignedPath.java
b/server/src/main/java/org/apache/iotdb/db/metadata/path/AlignedPath.java
index 44457ad..7a8ee70 100644
--- a/server/src/main/java/org/apache/iotdb/db/metadata/path/AlignedPath.java
+++ b/server/src/main/java/org/apache/iotdb/db/metadata/path/AlignedPath.java
@@ -139,10 +139,14 @@ public class AlignedPath extends PartialPath {
this.measurementList = measurementList;
}
- public void addMeasurement(List<String> measurements) {
+ public void addMeasurements(List<String> measurements) {
this.measurementList.addAll(measurements);
}
+ public void addSchemas(List<IMeasurementSchema> schemas) {
+ this.schemaList.addAll(schemas);
+ }
+
public void addMeasurement(MeasurementPath measurementPath) {
if (measurementList == null) {
measurementList = new ArrayList<>();
diff --git
a/server/src/main/java/org/apache/iotdb/db/query/aggregation/impl/CountAggrResult.java
b/server/src/main/java/org/apache/iotdb/db/query/aggregation/impl/CountAggrResult.java
index 10b9ad0..57a3a7e 100644
---
a/server/src/main/java/org/apache/iotdb/db/query/aggregation/impl/CountAggrResult.java
+++
b/server/src/main/java/org/apache/iotdb/db/query/aggregation/impl/CountAggrResult.java
@@ -57,6 +57,7 @@ public class CountAggrResult extends AggregateResult {
public void updateResultFromPageData(
IBatchDataIterator batchIterator, long minBound, long maxBound) {
int cnt = 0;
+ int count = batchIterator.totalLength();
while (batchIterator.hasNext()) {
if (batchIterator.currentTime() >= maxBound ||
batchIterator.currentTime() < minBound) {
break;
diff --git
a/server/src/main/java/org/apache/iotdb/db/query/executor/AggregationExecutor.java
b/server/src/main/java/org/apache/iotdb/db/query/executor/AggregationExecutor.java
index 3db98b7..9d622af 100644
---
a/server/src/main/java/org/apache/iotdb/db/query/executor/AggregationExecutor.java
+++
b/server/src/main/java/org/apache/iotdb/db/query/executor/AggregationExecutor.java
@@ -27,6 +27,7 @@ import
org.apache.iotdb.db.engine.storagegroup.StorageGroupProcessor;
import org.apache.iotdb.db.exception.StorageEngineException;
import org.apache.iotdb.db.exception.query.QueryProcessException;
import org.apache.iotdb.db.metadata.path.AlignedPath;
+import org.apache.iotdb.db.metadata.path.MeasurementPath;
import org.apache.iotdb.db.metadata.path.PartialPath;
import org.apache.iotdb.db.qp.physical.crud.AggregationPlan;
import org.apache.iotdb.db.qp.physical.crud.QueryPlan;
@@ -401,7 +402,7 @@ public class AggregationExecutor {
while (seriesReader.hasNextFile()) {
// cal by file statistics
- if (seriesReader.canUseCurrentFileStatistics()) {
+ if (seriesReader.canUseCurrentTimeFileStatistics()) {
while (seriesReader.hasNextSubSeries()) {
Statistics fileStatistics = seriesReader.currentFileStatistics();
remainingToCalculate =
@@ -422,7 +423,7 @@ public class AggregationExecutor {
while (seriesReader.hasNextChunk()) {
// cal by chunk statistics
- if (seriesReader.canUseCurrentChunkStatistics()) {
+ if (seriesReader.canUseCurrentTimeChunkStatistics()) {
while (seriesReader.hasNextSubSeries()) {
Statistics chunkStatistics = seriesReader.currentChunkStatistics();
remainingToCalculate =
@@ -515,7 +516,7 @@ public class AggregationExecutor {
throws IOException, QueryProcessException {
while (seriesReader.hasNextPage()) {
// cal by page statistics
- if (seriesReader.canUseCurrentPageStatistics()) {
+ if (seriesReader.canUseCurrentTimePageStatistics()) {
while (seriesReader.hasNextSubSeries()) {
Statistics pageStatistic = seriesReader.currentPageStatistics();
remainingToCalculate =
@@ -754,18 +755,21 @@ public class AggregationExecutor {
List<PartialPath> seriesPaths = new
ArrayList<>(pathToAggrIndexesMap.keySet());
for (PartialPath seriesPath : seriesPaths) {
- if (seriesPath instanceof AlignedPath) {
+ if (seriesPath instanceof MeasurementPath
+ && ((MeasurementPath) seriesPath).isUnderAlignedEntity()) {
List<Integer> indexes = pathToAggrIndexesMap.remove(seriesPath);
- AlignedPath groupPath = temp.get(seriesPath.getFullPath());
+ AlignedPath exactPath = (AlignedPath) ((MeasurementPath)
seriesPath).transformToExactPath();
+ AlignedPath groupPath = temp.get(exactPath.getFullPath());
if (groupPath == null) {
- groupPath = (AlignedPath) seriesPath.copy();
- temp.put(seriesPath.getFullPath(), groupPath);
+ groupPath = exactPath;
+ temp.put(groupPath.getFullPath(), groupPath);
result.computeIfAbsent(groupPath, key -> new
ArrayList<>()).add(indexes);
} else {
// groupPath is changed here so we update it
List<List<Integer>> subIndexes = result.remove(groupPath);
subIndexes.add(indexes);
- groupPath.addMeasurement(((AlignedPath)
seriesPath).getMeasurementList());
+ groupPath.addMeasurements(exactPath.getMeasurementList());
+ groupPath.addSchemas(exactPath.getSchemaList());
result.put(groupPath, subIndexes);
}
}
diff --git
a/server/src/main/java/org/apache/iotdb/db/query/reader/series/VectorSeriesAggregateReader.java
b/server/src/main/java/org/apache/iotdb/db/query/reader/series/VectorSeriesAggregateReader.java
index 29cb347..719149d 100644
---
a/server/src/main/java/org/apache/iotdb/db/query/reader/series/VectorSeriesAggregateReader.java
+++
b/server/src/main/java/org/apache/iotdb/db/query/reader/series/VectorSeriesAggregateReader.java
@@ -84,6 +84,13 @@ public class VectorSeriesAggregateReader implements
IAggregateReader {
&& !seriesReader.currentFileModified();
}
+ public boolean canUseCurrentTimeFileStatistics() throws IOException {
+ Statistics fileStatistics = seriesReader.currentFileStatistics();
+ return !seriesReader.isFileOverlapped()
+ && containedByTimeFilter(fileStatistics)
+ && !seriesReader.currentFileModified();
+ }
+
@Override
public Statistics currentFileStatistics() throws IOException {
return seriesReader.currentFileStatistics(curIndex);
@@ -107,6 +114,13 @@ public class VectorSeriesAggregateReader implements
IAggregateReader {
&& !seriesReader.currentChunkModified();
}
+ public boolean canUseCurrentTimeChunkStatistics() throws IOException {
+ Statistics chunkStatistics = seriesReader.currentChunkStatistics();
+ return !seriesReader.isChunkOverlapped()
+ && containedByTimeFilter(chunkStatistics)
+ && !seriesReader.currentChunkModified();
+ }
+
@Override
public Statistics currentChunkStatistics() throws IOException {
return seriesReader.currentChunkStatistics(curIndex);
@@ -133,6 +147,16 @@ public class VectorSeriesAggregateReader implements
IAggregateReader {
&& !seriesReader.currentPageModified();
}
+ public boolean canUseCurrentTimePageStatistics() throws IOException {
+ Statistics currentPageStatistics = seriesReader.currentPageStatistics();
+ if (currentPageStatistics == null) {
+ return false;
+ }
+ return !seriesReader.isPageOverlapped()
+ && containedByTimeFilter(currentPageStatistics)
+ && !seriesReader.currentPageModified();
+ }
+
@Override
public Statistics currentPageStatistics() throws IOException {
return seriesReader.currentPageStatistics(curIndex);
diff --git
a/server/src/test/java/org/apache/iotdb/db/integration/aligned/IoTDBAggregationWithoutValueFilter2IT.java
b/server/src/test/java/org/apache/iotdb/db/integration/aligned/IoTDBAggregationWithoutValueFilter2IT.java
new file mode 100644
index 0000000..1c799e0
--- /dev/null
+++
b/server/src/test/java/org/apache/iotdb/db/integration/aligned/IoTDBAggregationWithoutValueFilter2IT.java
@@ -0,0 +1,63 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iotdb.db.integration.aligned;
+
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.utils.EnvironmentUtils;
+import org.apache.iotdb.tsfile.common.conf.TSFileDescriptor;
+
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+
+/** Let One chunk has more than one page. */
+public class IoTDBAggregationWithoutValueFilter2IT extends
IoTDBAggregationWithoutValueFilterIT {
+ private static int numOfPointsPerPage;
+
+ @BeforeClass
+ public static void setUp() throws Exception {
+ EnvironmentUtils.closeStatMonitor();
+ EnvironmentUtils.envSetUp();
+ // TODO When the aligned time series support compaction, we need to set
compaction to true
+ enableSeqSpaceCompaction =
+ IoTDBDescriptor.getInstance().getConfig().isEnableSeqSpaceCompaction();
+ enableUnseqSpaceCompaction =
+
IoTDBDescriptor.getInstance().getConfig().isEnableUnseqSpaceCompaction();
+ enableCrossSpaceCompaction =
+
IoTDBDescriptor.getInstance().getConfig().isEnableCrossSpaceCompaction();
+ numOfPointsPerPage =
TSFileDescriptor.getInstance().getConfig().getMaxNumberOfPointsInPage();
+
IoTDBDescriptor.getInstance().getConfig().setEnableSeqSpaceCompaction(false);
+
IoTDBDescriptor.getInstance().getConfig().setEnableUnseqSpaceCompaction(false);
+
IoTDBDescriptor.getInstance().getConfig().setEnableCrossSpaceCompaction(false);
+ TSFileDescriptor.getInstance().getConfig().setMaxNumberOfPointsInPage(3);
+ AlignedWriteUtil.insertData();
+ }
+
+ @AfterClass
+ public static void tearDown() throws Exception {
+
IoTDBDescriptor.getInstance().getConfig().setEnableSeqSpaceCompaction(enableSeqSpaceCompaction);
+ IoTDBDescriptor.getInstance()
+ .getConfig()
+ .setEnableUnseqSpaceCompaction(enableUnseqSpaceCompaction);
+ IoTDBDescriptor.getInstance()
+ .getConfig()
+ .setEnableCrossSpaceCompaction(enableCrossSpaceCompaction);
+
TSFileDescriptor.getInstance().getConfig().setMaxNumberOfPointsInPage(numOfPointsPerPage);
+ EnvironmentUtils.cleanEnv();
+ }
+}
diff --git
a/server/src/test/java/org/apache/iotdb/db/integration/aligned/IoTDBAggregationWithoutValueFilterIT.java
b/server/src/test/java/org/apache/iotdb/db/integration/aligned/IoTDBAggregationWithoutValueFilterIT.java
new file mode 100644
index 0000000..2626105
--- /dev/null
+++
b/server/src/test/java/org/apache/iotdb/db/integration/aligned/IoTDBAggregationWithoutValueFilterIT.java
@@ -0,0 +1,444 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iotdb.db.integration.aligned;
+
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.utils.EnvironmentUtils;
+import org.apache.iotdb.jdbc.Config;
+
+import org.junit.*;
+
+import java.sql.*;
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.junit.Assert.*;
+
+public class IoTDBAggregationWithoutValueFilterIT {
+
+ private static final double DELTA = 1e-6;
+ protected static boolean enableSeqSpaceCompaction;
+ protected static boolean enableUnseqSpaceCompaction;
+ protected static boolean enableCrossSpaceCompaction;
+
+ @BeforeClass
+ public static void setUp() throws Exception {
+ EnvironmentUtils.closeStatMonitor();
+ EnvironmentUtils.envSetUp();
+ // TODO When the aligned time series support compaction, we need to set
compaction to true
+ enableSeqSpaceCompaction =
+ IoTDBDescriptor.getInstance().getConfig().isEnableSeqSpaceCompaction();
+ enableUnseqSpaceCompaction =
+
IoTDBDescriptor.getInstance().getConfig().isEnableUnseqSpaceCompaction();
+ enableCrossSpaceCompaction =
+
IoTDBDescriptor.getInstance().getConfig().isEnableCrossSpaceCompaction();
+
IoTDBDescriptor.getInstance().getConfig().setEnableSeqSpaceCompaction(false);
+
IoTDBDescriptor.getInstance().getConfig().setEnableUnseqSpaceCompaction(false);
+
IoTDBDescriptor.getInstance().getConfig().setEnableCrossSpaceCompaction(false);
+ AlignedWriteUtil.insertData();
+ }
+
+ @AfterClass
+ public static void tearDown() throws Exception {
+
IoTDBDescriptor.getInstance().getConfig().setEnableSeqSpaceCompaction(enableSeqSpaceCompaction);
+ IoTDBDescriptor.getInstance()
+ .getConfig()
+ .setEnableUnseqSpaceCompaction(enableUnseqSpaceCompaction);
+ IoTDBDescriptor.getInstance()
+ .getConfig()
+ .setEnableCrossSpaceCompaction(enableCrossSpaceCompaction);
+ EnvironmentUtils.cleanEnv();
+ }
+
+ @Test
+ public void countAllAlignedWithoutTimeFilterTest() throws
ClassNotFoundException {
+ String[] retArray = new String[] {"20", "29", "28", "19", "20"};
+ String[] columnNames = {
+ "count(root.sg1.d1.s1)",
+ "count(root.sg1.d1.s2)",
+ "count(root.sg1.d1.s3)",
+ "count(root.sg1.d1.s4)",
+ "count(root.sg1.d1.s5)"
+ };
+ Class.forName(Config.JDBC_DRIVER_NAME);
+ try (Connection connection =
+ DriverManager.getConnection(
+ Config.IOTDB_URL_PREFIX + "127.0.0.1:6667/", "root", "root");
+ Statement statement = connection.createStatement()) {
+
+ boolean hasResultSet = statement.execute("select count(*) from
root.sg1.d1");
+ Assert.assertTrue(hasResultSet);
+ try (ResultSet resultSet = statement.getResultSet()) {
+ ResultSetMetaData resultSetMetaData = resultSet.getMetaData();
+ Map<String, Integer> map = new HashMap<>(); // used to adjust result
sequence
+ for (int i = 1; i <= resultSetMetaData.getColumnCount(); i++) {
+ map.put(resultSetMetaData.getColumnName(i), i);
+ }
+ assertEquals(columnNames.length, resultSetMetaData.getColumnCount());
+ int cnt = 0;
+ while (resultSet.next()) {
+ String[] ans = new String[columnNames.length];
+ // No need to add time column for aggregation query
+ for (int i = 0; i < columnNames.length; i++) {
+ String columnName = columnNames[i];
+ int index = map.get(columnName);
+ ans[i] = resultSet.getString(index);
+ }
+ assertArrayEquals(retArray, ans);
+ cnt++;
+ }
+ assertEquals(1, cnt);
+ }
+ } catch (SQLException e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ @Test
+ public void countAllAlignedAndNonAlignedWithoutTimeFilterTest() throws
ClassNotFoundException {
+ String[] retArray = new String[] {"20", "29", "28", "19", "20", "19",
"29", "28", "18", "19"};
+ String[] columnNames = {
+ "count(root.sg1.d1.s1)",
+ "count(root.sg1.d1.s2)",
+ "count(root.sg1.d1.s3)",
+ "count(root.sg1.d1.s4)",
+ "count(root.sg1.d1.s5)",
+ "count(root.sg1.d2.s1)",
+ "count(root.sg1.d2.s2)",
+ "count(root.sg1.d2.s3)",
+ "count(root.sg1.d2.s4)",
+ "count(root.sg1.d2.s5)"
+ };
+ Class.forName(Config.JDBC_DRIVER_NAME);
+ try (Connection connection =
+ DriverManager.getConnection(
+ Config.IOTDB_URL_PREFIX + "127.0.0.1:6667/", "root", "root");
+ Statement statement = connection.createStatement()) {
+
+ boolean hasResultSet = statement.execute("select count(*) from
root.sg1.*");
+ Assert.assertTrue(hasResultSet);
+ try (ResultSet resultSet = statement.getResultSet()) {
+ ResultSetMetaData resultSetMetaData = resultSet.getMetaData();
+ Map<String, Integer> map = new HashMap<>();
+ for (int i = 1; i <= resultSetMetaData.getColumnCount(); i++) {
+ map.put(resultSetMetaData.getColumnName(i), i);
+ }
+ assertEquals(columnNames.length, resultSetMetaData.getColumnCount());
+ int cnt = 0;
+ while (resultSet.next()) {
+ String[] ans = new String[columnNames.length];
+ // No need to add time column for aggregation query
+ for (int i = 0; i < columnNames.length; i++) {
+ String columnName = columnNames[i];
+ int index = map.get(columnName);
+ ans[i] = resultSet.getString(index);
+ }
+ assertArrayEquals(retArray, ans);
+ cnt++;
+ }
+ assertEquals(1, cnt);
+ }
+ } catch (SQLException e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ @Test
+ public void countAllAlignedWithTimeFilterTest() throws
ClassNotFoundException {
+ String[] retArray = new String[] {"12", "15", "22", "13", "6"};
+ String[] columnNames = {
+ "count(root.sg1.d1.s1)",
+ "count(root.sg1.d1.s2)",
+ "count(root.sg1.d1.s3)",
+ "count(root.sg1.d1.s4)",
+ "count(root.sg1.d1.s5)"
+ };
+ Class.forName(Config.JDBC_DRIVER_NAME);
+ try (Connection connection =
+ DriverManager.getConnection(
+ Config.IOTDB_URL_PREFIX + "127.0.0.1:6667/", "root", "root");
+ Statement statement = connection.createStatement()) {
+
+ boolean hasResultSet =
+ statement.execute("select count(*) from root.sg1.d1 where time >= 9
and time <= 33");
+ Assert.assertTrue(hasResultSet);
+ try (ResultSet resultSet = statement.getResultSet()) {
+ ResultSetMetaData resultSetMetaData = resultSet.getMetaData();
+ Map<String, Integer> map = new HashMap<>();
+ for (int i = 1; i <= resultSetMetaData.getColumnCount(); i++) {
+ map.put(resultSetMetaData.getColumnName(i), i);
+ }
+ assertEquals(columnNames.length, resultSetMetaData.getColumnCount());
+ int cnt = 0;
+ while (resultSet.next()) {
+ String[] ans = new String[columnNames.length];
+ // No need to add time column for aggregation query
+ for (int i = 0; i < columnNames.length; i++) {
+ String columnName = columnNames[i];
+ int index = map.get(columnName);
+ ans[i] = resultSet.getString(index);
+ }
+ assertArrayEquals(retArray, ans);
+ cnt++;
+ }
+ assertEquals(1, cnt);
+ }
+ } catch (SQLException e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ /** aggregate multi columns of aligned timeseries in one SQL */
+ @Test
+ public void aggregateSomeAlignedWithoutTimeFilterTest() throws
ClassNotFoundException {
+ double[] retArray =
+ new double[] {
+ 20, 29, 28, 390184, 130549, 390417, 19509.2, 4501.689655172413,
13943.464285714286
+ };
+ String[] columnNames = {
+ "count(root.sg1.d1.s1)",
+ "count(root.sg1.d1.s2)",
+ "count(root.sg1.d1.s3)",
+ "sum(root.sg1.d1.s1)",
+ "sum(root.sg1.d1.s2)",
+ "sum(root.sg1.d1.s3)",
+ "avg(root.sg1.d1.s1)",
+ "avg(root.sg1.d1.s2)",
+ "avg(root.sg1.d1.s3)",
+ };
+ Class.forName(Config.JDBC_DRIVER_NAME);
+ try (Connection connection =
+ DriverManager.getConnection(
+ Config.IOTDB_URL_PREFIX + "127.0.0.1:6667/", "root", "root");
+ Statement statement = connection.createStatement()) {
+
+ boolean hasResultSet =
+ statement.execute(
+ "select count(s1),count (s2),count
(s3),sum(s1),sum(s2),sum(s3),avg(s1),avg(s2),avg(s3) from root.sg1.d1");
+ Assert.assertTrue(hasResultSet);
+ try (ResultSet resultSet = statement.getResultSet()) {
+ ResultSetMetaData resultSetMetaData = resultSet.getMetaData();
+ Map<String, Integer> map = new HashMap<>();
+ for (int i = 1; i <= resultSetMetaData.getColumnCount(); i++) {
+ map.put(resultSetMetaData.getColumnName(i), i);
+ }
+ assertEquals(columnNames.length, resultSetMetaData.getColumnCount());
+ int cnt = 0;
+ while (resultSet.next()) {
+ double[] ans = new double[columnNames.length];
+ // No need to add time column for aggregation query
+ for (int i = 0; i < columnNames.length; i++) {
+ String columnName = columnNames[i];
+ int index = map.get(columnName);
+ ans[i] = Double.parseDouble(resultSet.getString(index));
+ }
+ assertArrayEquals(retArray, ans, DELTA);
+ cnt++;
+ }
+ assertEquals(1, cnt);
+ }
+ } catch (SQLException e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ /** aggregate multi columns of aligned timeseries in one SQL */
+ @Test
+ public void aggregateSomeAlignedWithTimeFilterTest() throws
ClassNotFoundException {
+ double[] retArray =
+ new double[] {
+ 6, 9, 15, 230090, 220, 230322, 38348.333333333336,
24.444444444444443, 15354.8
+ };
+ String[] columnNames = {
+ "count(root.sg1.d1.s1)",
+ "count(root.sg1.d1.s2)",
+ "count(root.sg1.d1.s3)",
+ "sum(root.sg1.d1.s1)",
+ "sum(root.sg1.d1.s2)",
+ "sum(root.sg1.d1.s3)",
+ "avg(root.sg1.d1.s1)",
+ "avg(root.sg1.d1.s2)",
+ "avg(root.sg1.d1.s3)",
+ };
+ Class.forName(Config.JDBC_DRIVER_NAME);
+ try (Connection connection =
+ DriverManager.getConnection(
+ Config.IOTDB_URL_PREFIX + "127.0.0.1:6667/", "root", "root");
+ Statement statement = connection.createStatement()) {
+
+ boolean hasResultSet =
+ statement.execute(
+ "select count(s1),count (s2),count
(s3),sum(s1),sum(s2),sum(s3),avg(s1),avg(s2),avg(s3) from root.sg1.d1 where
time>=16 and time<=34");
+ Assert.assertTrue(hasResultSet);
+ try (ResultSet resultSet = statement.getResultSet()) {
+ ResultSetMetaData resultSetMetaData = resultSet.getMetaData();
+ Map<String, Integer> map = new HashMap<>();
+ for (int i = 1; i <= resultSetMetaData.getColumnCount(); i++) {
+ map.put(resultSetMetaData.getColumnName(i), i);
+ }
+ assertEquals(columnNames.length, resultSetMetaData.getColumnCount());
+ int cnt = 0;
+ while (resultSet.next()) {
+ double[] ans = new double[columnNames.length];
+ // No need to add time column for aggregation query
+ for (int i = 0; i < columnNames.length; i++) {
+ String columnName = columnNames[i];
+ int index = map.get(columnName);
+ ans[i] = Double.parseDouble(resultSet.getString(index));
+ }
+ assertArrayEquals(retArray, ans, DELTA);
+ cnt++;
+ }
+ assertEquals(1, cnt);
+ }
+ } catch (SQLException e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ @Test
+ public void countSingleAlignedWithTimeFilterTest() throws
ClassNotFoundException {
+ String[] retArray = new String[] {"9"};
+ String[] columnNames = {"count(root.sg1.d1.s2)"};
+ Class.forName(Config.JDBC_DRIVER_NAME);
+ try (Connection connection =
+ DriverManager.getConnection(
+ Config.IOTDB_URL_PREFIX + "127.0.0.1:6667/", "root", "root");
+ Statement statement = connection.createStatement()) {
+
+ boolean hasResultSet =
+ statement.execute("select count(s2) from root.sg1.d1 where time>=16
and time<=34");
+ Assert.assertTrue(hasResultSet);
+ try (ResultSet resultSet = statement.getResultSet()) {
+ ResultSetMetaData resultSetMetaData = resultSet.getMetaData();
+ Map<String, Integer> map = new HashMap<>();
+ for (int i = 1; i <= resultSetMetaData.getColumnCount(); i++) {
+ map.put(resultSetMetaData.getColumnName(i), i);
+ }
+ assertEquals(columnNames.length, resultSetMetaData.getColumnCount());
+ int cnt = 0;
+ while (resultSet.next()) {
+ StringBuilder builder = new StringBuilder();
+ // No need to add time column for aggregation query
+ for (String columnName : columnNames) {
+ int index = map.get(columnName);
+ if (builder.length() != 0) {
+ builder.append(",");
+ }
+ builder.append(resultSet.getString(index));
+ }
+ assertEquals(retArray[cnt], builder.toString());
+ cnt++;
+ }
+ assertEquals(1, cnt);
+ }
+ } catch (SQLException e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ @Test
+ public void sumSingleAlignedWithTimeFilterTest() throws
ClassNotFoundException {
+ String[] retArray = new String[] {"230322.0"};
+ String[] columnNames = {"sum(root.sg1.d1.s3)"};
+ Class.forName(Config.JDBC_DRIVER_NAME);
+ try (Connection connection =
+ DriverManager.getConnection(
+ Config.IOTDB_URL_PREFIX + "127.0.0.1:6667/", "root", "root");
+ Statement statement = connection.createStatement()) {
+
+ boolean hasResultSet =
+ statement.execute("select sum(s3) from root.sg1.d1 where time>=16
and time<=34");
+ Assert.assertTrue(hasResultSet);
+ try (ResultSet resultSet = statement.getResultSet()) {
+ ResultSetMetaData resultSetMetaData = resultSet.getMetaData();
+ Map<String, Integer> map = new HashMap<>();
+ for (int i = 1; i <= resultSetMetaData.getColumnCount(); i++) {
+ map.put(resultSetMetaData.getColumnName(i), i);
+ }
+ assertEquals(columnNames.length, resultSetMetaData.getColumnCount());
+ int cnt = 0;
+ while (resultSet.next()) {
+ StringBuilder builder = new StringBuilder();
+ // No need to add time column for aggregation query
+ for (String columnName : columnNames) {
+ int index = map.get(columnName);
+ if (builder.length() != 0) {
+ builder.append(",");
+ }
+ builder.append(resultSet.getString(index));
+ }
+ assertEquals(retArray[cnt], builder.toString());
+ cnt++;
+ }
+ assertEquals(1, cnt);
+ }
+ } catch (SQLException e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ @Test
+ public void avgSingleAlignedWithTimeFilterTest() throws
ClassNotFoundException {
+ double[][] retArray = {{24.444444444444443}};
+ String[] columnNames = {"avg(root.sg1.d1.s2)"};
+ Class.forName(Config.JDBC_DRIVER_NAME);
+ try (Connection connection =
+ DriverManager.getConnection(
+ Config.IOTDB_URL_PREFIX + "127.0.0.1:6667/", "root", "root");
+ Statement statement = connection.createStatement()) {
+
+ boolean hasResultSet =
+ statement.execute("select avg(s2) from root.sg1.d1 where time>=16
and time<=34");
+ Assert.assertTrue(hasResultSet);
+ try (ResultSet resultSet = statement.getResultSet()) {
+ ResultSetMetaData resultSetMetaData = resultSet.getMetaData();
+ Map<String, Integer> map = new HashMap<>();
+ for (int i = 1; i <= resultSetMetaData.getColumnCount(); i++) {
+ map.put(resultSetMetaData.getColumnName(i), i);
+ }
+ assertEquals(columnNames.length, resultSetMetaData.getColumnCount());
+ int cnt = 0;
+ while (resultSet.next()) {
+ double[] ans = new double[columnNames.length];
+ StringBuilder builder = new StringBuilder();
+ // No need to add time column for aggregation query
+ for (int i = 0; i < columnNames.length; i++) {
+ String columnName = columnNames[i];
+ int index = map.get(columnName);
+ ans[i] = Double.parseDouble(resultSet.getString(index));
+ }
+ assertArrayEquals(retArray[cnt], ans, DELTA);
+ cnt++;
+ }
+ assertEquals(1, cnt);
+ }
+ } catch (SQLException e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+}
diff --git
a/server/src/test/java/org/apache/iotdb/db/integration/aligned/IoTDBAggregationWithoutValueFilterWithDeletion2IT.java
b/server/src/test/java/org/apache/iotdb/db/integration/aligned/IoTDBAggregationWithoutValueFilterWithDeletion2IT.java
new file mode 100644
index 0000000..40f14e4
--- /dev/null
+++
b/server/src/test/java/org/apache/iotdb/db/integration/aligned/IoTDBAggregationWithoutValueFilterWithDeletion2IT.java
@@ -0,0 +1,83 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iotdb.db.integration.aligned;
+
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.utils.EnvironmentUtils;
+import org.apache.iotdb.jdbc.Config;
+import org.apache.iotdb.tsfile.common.conf.TSFileDescriptor;
+
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.Statement;
+
+/** Let One chunk has more than one page. */
+public class IoTDBAggregationWithoutValueFilterWithDeletion2IT
+ extends IoTDBAggregationWithoutValueFilterWithDeletionIT {
+
+ private static int numOfPointsPerPage;
+
+ @BeforeClass
+ public static void setUp() throws Exception {
+ EnvironmentUtils.closeStatMonitor();
+ EnvironmentUtils.envSetUp();
+
+ enableSeqSpaceCompaction =
+ IoTDBDescriptor.getInstance().getConfig().isEnableSeqSpaceCompaction();
+ enableUnseqSpaceCompaction =
+
IoTDBDescriptor.getInstance().getConfig().isEnableUnseqSpaceCompaction();
+ enableCrossSpaceCompaction =
+
IoTDBDescriptor.getInstance().getConfig().isEnableCrossSpaceCompaction();
+ numOfPointsPerPage =
TSFileDescriptor.getInstance().getConfig().getMaxNumberOfPointsInPage();
+
IoTDBDescriptor.getInstance().getConfig().setEnableSeqSpaceCompaction(false);
+
IoTDBDescriptor.getInstance().getConfig().setEnableUnseqSpaceCompaction(false);
+
IoTDBDescriptor.getInstance().getConfig().setEnableCrossSpaceCompaction(false);
+ TSFileDescriptor.getInstance().getConfig().setMaxNumberOfPointsInPage(3);
+ AlignedWriteUtil.insertData();
+ Class.forName(Config.JDBC_DRIVER_NAME);
+ try (Connection connection =
+ DriverManager.getConnection(
+ Config.IOTDB_URL_PREFIX + "127.0.0.1:6667/", "root", "root");
+ Statement statement = connection.createStatement()) {
+ // TODO currently aligned data in memory doesn't support deletion, so we
flush all data to
+ // disk before doing deletion
+ statement.execute("flush");
+ statement.execute("delete timeseries root.sg1.d1.s2");
+ statement.execute("delete from root.sg1.d1.s1 where time <= 21");
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+ }
+
+ @AfterClass
+ public static void tearDown() throws Exception {
+
IoTDBDescriptor.getInstance().getConfig().setEnableSeqSpaceCompaction(enableSeqSpaceCompaction);
+ IoTDBDescriptor.getInstance()
+ .getConfig()
+ .setEnableUnseqSpaceCompaction(enableUnseqSpaceCompaction);
+ IoTDBDescriptor.getInstance()
+ .getConfig()
+ .setEnableCrossSpaceCompaction(enableCrossSpaceCompaction);
+
TSFileDescriptor.getInstance().getConfig().setMaxNumberOfPointsInPage(numOfPointsPerPage);
+ EnvironmentUtils.cleanEnv();
+ }
+}
diff --git
a/server/src/test/java/org/apache/iotdb/db/integration/aligned/IoTDBAggregationWithoutValueFilterWithDeletionIT.java
b/server/src/test/java/org/apache/iotdb/db/integration/aligned/IoTDBAggregationWithoutValueFilterWithDeletionIT.java
new file mode 100644
index 0000000..bf68c50
--- /dev/null
+++
b/server/src/test/java/org/apache/iotdb/db/integration/aligned/IoTDBAggregationWithoutValueFilterWithDeletionIT.java
@@ -0,0 +1,447 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iotdb.db.integration.aligned;
+
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.utils.EnvironmentUtils;
+import org.apache.iotdb.jdbc.Config;
+
+import org.junit.AfterClass;
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import java.sql.*;
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.junit.Assert.*;
+
+public class IoTDBAggregationWithoutValueFilterWithDeletionIT {
+
+ private static final double DELTA = 1e-6;
+ protected static boolean enableSeqSpaceCompaction;
+ protected static boolean enableUnseqSpaceCompaction;
+ protected static boolean enableCrossSpaceCompaction;
+
+ @BeforeClass
+ public static void setUp() throws Exception {
+ EnvironmentUtils.closeStatMonitor();
+ EnvironmentUtils.envSetUp();
+
+ enableSeqSpaceCompaction =
+ IoTDBDescriptor.getInstance().getConfig().isEnableSeqSpaceCompaction();
+ enableUnseqSpaceCompaction =
+
IoTDBDescriptor.getInstance().getConfig().isEnableUnseqSpaceCompaction();
+ enableCrossSpaceCompaction =
+
IoTDBDescriptor.getInstance().getConfig().isEnableCrossSpaceCompaction();
+
IoTDBDescriptor.getInstance().getConfig().setEnableSeqSpaceCompaction(false);
+
IoTDBDescriptor.getInstance().getConfig().setEnableUnseqSpaceCompaction(false);
+
IoTDBDescriptor.getInstance().getConfig().setEnableCrossSpaceCompaction(false);
+
+ AlignedWriteUtil.insertData();
+ Class.forName(Config.JDBC_DRIVER_NAME);
+ try (Connection connection =
+ DriverManager.getConnection(
+ Config.IOTDB_URL_PREFIX + "127.0.0.1:6667/", "root", "root");
+ Statement statement = connection.createStatement()) {
+ // TODO currently aligned data in memory doesn't support deletion, so we
flush all data to
+ // disk before doing deletion
+ statement.execute("flush");
+ statement.execute("delete timeseries root.sg1.d1.s2");
+ statement.execute("delete from root.sg1.d1.s1 where time <= 21");
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+ }
+
+ @AfterClass
+ public static void tearDown() throws Exception {
+
IoTDBDescriptor.getInstance().getConfig().setEnableSeqSpaceCompaction(enableSeqSpaceCompaction);
+ IoTDBDescriptor.getInstance()
+ .getConfig()
+ .setEnableUnseqSpaceCompaction(enableUnseqSpaceCompaction);
+ IoTDBDescriptor.getInstance()
+ .getConfig()
+ .setEnableCrossSpaceCompaction(enableCrossSpaceCompaction);
+ EnvironmentUtils.cleanEnv();
+ }
+
+ @Test
+ public void countAllAlignedWithoutTimeFilterTest() throws
ClassNotFoundException {
+ String[] retArray = new String[] {"1", "28", "19", "20"};
+ String[] columnNames = {
+ "count(root.sg1.d1.s1)",
+ "count(root.sg1.d1.s3)",
+ "count(root.sg1.d1.s4)",
+ "count(root.sg1.d1.s5)"
+ };
+ Class.forName(Config.JDBC_DRIVER_NAME);
+ try (Connection connection =
+ DriverManager.getConnection(
+ Config.IOTDB_URL_PREFIX + "127.0.0.1:6667/", "root", "root");
+ Statement statement = connection.createStatement()) {
+
+ boolean hasResultSet = statement.execute("select count(*) from
root.sg1.d1");
+ Assert.assertTrue(hasResultSet);
+ try (ResultSet resultSet = statement.getResultSet()) {
+ ResultSetMetaData resultSetMetaData = resultSet.getMetaData();
+ Map<String, Integer> map = new HashMap<>(); // used to adjust result
sequence
+ for (int i = 1; i <= resultSetMetaData.getColumnCount(); i++) {
+ map.put(resultSetMetaData.getColumnName(i), i);
+ }
+ assertEquals(columnNames.length, resultSetMetaData.getColumnCount());
+ int cnt = 0;
+ while (resultSet.next()) {
+ String[] ans = new String[columnNames.length];
+ // No need to add time column for aggregation query
+ for (int i = 0; i < columnNames.length; i++) {
+ String columnName = columnNames[i];
+ int index = map.get(columnName);
+ ans[i] = resultSet.getString(index);
+ }
+ assertArrayEquals(retArray, ans);
+ cnt++;
+ }
+ assertEquals(1, cnt);
+ }
+ } catch (SQLException e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ @Test
+ public void countAllAlignedAndNonAlignedWithoutTimeFilterTest() throws
ClassNotFoundException {
+ String[] retArray = new String[] {"1", "28", "19", "20", "19", "29", "28",
"18", "19"};
+ String[] columnNames = {
+ "count(root.sg1.d1.s1)",
+ "count(root.sg1.d1.s3)",
+ "count(root.sg1.d1.s4)",
+ "count(root.sg1.d1.s5)",
+ "count(root.sg1.d2.s1)",
+ "count(root.sg1.d2.s2)",
+ "count(root.sg1.d2.s3)",
+ "count(root.sg1.d2.s4)",
+ "count(root.sg1.d2.s5)"
+ };
+ Class.forName(Config.JDBC_DRIVER_NAME);
+ try (Connection connection =
+ DriverManager.getConnection(
+ Config.IOTDB_URL_PREFIX + "127.0.0.1:6667/", "root", "root");
+ Statement statement = connection.createStatement()) {
+
+ boolean hasResultSet = statement.execute("select count(*) from
root.sg1.*");
+ Assert.assertTrue(hasResultSet);
+ try (ResultSet resultSet = statement.getResultSet()) {
+ ResultSetMetaData resultSetMetaData = resultSet.getMetaData();
+ Map<String, Integer> map = new HashMap<>();
+ for (int i = 1; i <= resultSetMetaData.getColumnCount(); i++) {
+ map.put(resultSetMetaData.getColumnName(i), i);
+ }
+ assertEquals(columnNames.length, resultSetMetaData.getColumnCount());
+ int cnt = 0;
+ while (resultSet.next()) {
+ String[] ans = new String[columnNames.length];
+ // No need to add time column for aggregation query
+ for (int i = 0; i < columnNames.length; i++) {
+ String columnName = columnNames[i];
+ int index = map.get(columnName);
+ ans[i] = resultSet.getString(index);
+ }
+ assertArrayEquals(retArray, ans);
+ cnt++;
+ }
+ assertEquals(1, cnt);
+ }
+ } catch (SQLException e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ @Test
+ public void countAllAlignedWithTimeFilterTest() throws
ClassNotFoundException {
+ String[] retArray = new String[] {"1", "22", "13", "6"};
+ String[] columnNames = {
+ "count(root.sg1.d1.s1)",
+ "count(root.sg1.d1.s3)",
+ "count(root.sg1.d1.s4)",
+ "count(root.sg1.d1.s5)"
+ };
+ Class.forName(Config.JDBC_DRIVER_NAME);
+ try (Connection connection =
+ DriverManager.getConnection(
+ Config.IOTDB_URL_PREFIX + "127.0.0.1:6667/", "root", "root");
+ Statement statement = connection.createStatement()) {
+
+ boolean hasResultSet =
+ statement.execute("select count(*) from root.sg1.d1 where time >= 9
and time <= 33");
+ Assert.assertTrue(hasResultSet);
+ try (ResultSet resultSet = statement.getResultSet()) {
+ ResultSetMetaData resultSetMetaData = resultSet.getMetaData();
+ Map<String, Integer> map = new HashMap<>();
+ for (int i = 1; i <= resultSetMetaData.getColumnCount(); i++) {
+ map.put(resultSetMetaData.getColumnName(i), i);
+ }
+ assertEquals(columnNames.length, resultSetMetaData.getColumnCount());
+ int cnt = 0;
+ while (resultSet.next()) {
+ String[] ans = new String[columnNames.length];
+ // No need to add time column for aggregation query
+ for (int i = 0; i < columnNames.length; i++) {
+ String columnName = columnNames[i];
+ int index = map.get(columnName);
+ ans[i] = resultSet.getString(index);
+ }
+ assertArrayEquals(retArray, ans);
+ cnt++;
+ }
+ assertEquals(1, cnt);
+ }
+ } catch (SQLException e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ /** aggregate multi columns of aligned timeseries in one SQL */
+ @Test
+ public void aggregateSomeAlignedWithoutTimeFilterTest() throws
ClassNotFoundException {
+ double[] retArray = new double[] {1, 28, 230000, 390417, 230000,
13943.464285714286};
+ String[] columnNames = {
+ "count(root.sg1.d1.s1)",
+ "count(root.sg1.d1.s3)",
+ "sum(root.sg1.d1.s1)",
+ "sum(root.sg1.d1.s3)",
+ "avg(root.sg1.d1.s1)",
+ "avg(root.sg1.d1.s3)",
+ };
+ Class.forName(Config.JDBC_DRIVER_NAME);
+ try (Connection connection =
+ DriverManager.getConnection(
+ Config.IOTDB_URL_PREFIX + "127.0.0.1:6667/", "root", "root");
+ Statement statement = connection.createStatement()) {
+
+ boolean hasResultSet =
+ statement.execute(
+ "select count(s1),count (s3),sum(s1),sum(s3),avg(s1),avg(s3)
from root.sg1.d1");
+ Assert.assertTrue(hasResultSet);
+ try (ResultSet resultSet = statement.getResultSet()) {
+ ResultSetMetaData resultSetMetaData = resultSet.getMetaData();
+ Map<String, Integer> map = new HashMap<>();
+ for (int i = 1; i <= resultSetMetaData.getColumnCount(); i++) {
+ map.put(resultSetMetaData.getColumnName(i), i);
+ }
+ assertEquals(columnNames.length, resultSetMetaData.getColumnCount());
+ int cnt = 0;
+ while (resultSet.next()) {
+ double[] ans = new double[columnNames.length];
+ // No need to add time column for aggregation query
+ for (int i = 0; i < columnNames.length; i++) {
+ String columnName = columnNames[i];
+ int index = map.get(columnName);
+ ans[i] = Double.parseDouble(resultSet.getString(index));
+ }
+ assertArrayEquals(retArray, ans, DELTA);
+ cnt++;
+ }
+ assertEquals(1, cnt);
+ }
+ } catch (SQLException e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ /** aggregate multi columns of aligned timeseries in one SQL */
+ @Test
+ public void aggregateSomeAlignedWithTimeFilterTest() throws
ClassNotFoundException {
+ double[] retArray = new double[] {1, 15, 230000, 230322, 230000, 15354.8};
+ String[] columnNames = {
+ "count(root.sg1.d1.s1)",
+ "count(root.sg1.d1.s3)",
+ "sum(root.sg1.d1.s1)",
+ "sum(root.sg1.d1.s3)",
+ "avg(root.sg1.d1.s1)",
+ "avg(root.sg1.d1.s3)",
+ };
+ Class.forName(Config.JDBC_DRIVER_NAME);
+ try (Connection connection =
+ DriverManager.getConnection(
+ Config.IOTDB_URL_PREFIX + "127.0.0.1:6667/", "root", "root");
+ Statement statement = connection.createStatement()) {
+
+ boolean hasResultSet =
+ statement.execute(
+ "select count(s1),count (s3),sum(s1),sum(s3),avg(s1),avg(s3)
from root.sg1.d1 where time>=16 and time<=34");
+ Assert.assertTrue(hasResultSet);
+ try (ResultSet resultSet = statement.getResultSet()) {
+ ResultSetMetaData resultSetMetaData = resultSet.getMetaData();
+ Map<String, Integer> map = new HashMap<>();
+ for (int i = 1; i <= resultSetMetaData.getColumnCount(); i++) {
+ map.put(resultSetMetaData.getColumnName(i), i);
+ }
+ assertEquals(columnNames.length, resultSetMetaData.getColumnCount());
+ int cnt = 0;
+ while (resultSet.next()) {
+ double[] ans = new double[columnNames.length];
+ // No need to add time column for aggregation query
+ for (int i = 0; i < columnNames.length; i++) {
+ String columnName = columnNames[i];
+ int index = map.get(columnName);
+ ans[i] = Double.parseDouble(resultSet.getString(index));
+ }
+ assertArrayEquals(retArray, ans, DELTA);
+ cnt++;
+ }
+ assertEquals(1, cnt);
+ }
+ } catch (SQLException e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ @Test
+ public void countSingleAlignedWithTimeFilterTest() throws
ClassNotFoundException {
+ String[] retArray = new String[] {"0"};
+ String[] columnNames = {"count(root.sg1.d1.s1)"};
+ Class.forName(Config.JDBC_DRIVER_NAME);
+ try (Connection connection =
+ DriverManager.getConnection(
+ Config.IOTDB_URL_PREFIX + "127.0.0.1:6667/", "root", "root");
+ Statement statement = connection.createStatement()) {
+
+ boolean hasResultSet =
+ statement.execute("select count(s1) from root.sg1.d1 where time>=16
and time<=20");
+ Assert.assertTrue(hasResultSet);
+ try (ResultSet resultSet = statement.getResultSet()) {
+ ResultSetMetaData resultSetMetaData = resultSet.getMetaData();
+ Map<String, Integer> map = new HashMap<>();
+ for (int i = 1; i <= resultSetMetaData.getColumnCount(); i++) {
+ map.put(resultSetMetaData.getColumnName(i), i);
+ }
+ assertEquals(columnNames.length, resultSetMetaData.getColumnCount());
+ int cnt = 0;
+ while (resultSet.next()) {
+ StringBuilder builder = new StringBuilder();
+ // No need to add time column for aggregation query
+ for (String columnName : columnNames) {
+ int index = map.get(columnName);
+ if (builder.length() != 0) {
+ builder.append(",");
+ }
+ builder.append(resultSet.getString(index));
+ }
+ assertEquals(retArray[cnt], builder.toString());
+ cnt++;
+ }
+ assertEquals(1, cnt);
+ }
+ } catch (SQLException e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ @Test
+ public void sumSingleAlignedWithTimeFilterTest() throws
ClassNotFoundException {
+ String[] retArray = new String[] {"null"};
+ String[] columnNames = {"sum(root.sg1.d1.s1)"};
+ Class.forName(Config.JDBC_DRIVER_NAME);
+ try (Connection connection =
+ DriverManager.getConnection(
+ Config.IOTDB_URL_PREFIX + "127.0.0.1:6667/", "root", "root");
+ Statement statement = connection.createStatement()) {
+
+ boolean hasResultSet =
+ statement.execute("select sum(s1) from root.sg1.d1 where time>=16
and time<=20");
+ Assert.assertTrue(hasResultSet);
+ try (ResultSet resultSet = statement.getResultSet()) {
+ ResultSetMetaData resultSetMetaData = resultSet.getMetaData();
+ Map<String, Integer> map = new HashMap<>();
+ for (int i = 1; i <= resultSetMetaData.getColumnCount(); i++) {
+ map.put(resultSetMetaData.getColumnName(i), i);
+ }
+ assertEquals(columnNames.length, resultSetMetaData.getColumnCount());
+ int cnt = 0;
+ while (resultSet.next()) {
+ StringBuilder builder = new StringBuilder();
+ // No need to add time column for aggregation query
+ for (String columnName : columnNames) {
+ int index = map.get(columnName);
+ if (builder.length() != 0) {
+ builder.append(",");
+ }
+ builder.append(resultSet.getString(index));
+ }
+ assertEquals(retArray[cnt], builder.toString());
+ cnt++;
+ }
+ assertEquals(1, cnt);
+ }
+ } catch (SQLException e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ @Test
+ public void avgSingleAlignedWithTimeFilterTest() throws
ClassNotFoundException {
+ String[] retArray = new String[] {"null"};
+ String[] columnNames = {"avg(root.sg1.d1.s1)"};
+ Class.forName(Config.JDBC_DRIVER_NAME);
+ try (Connection connection =
+ DriverManager.getConnection(
+ Config.IOTDB_URL_PREFIX + "127.0.0.1:6667/", "root", "root");
+ Statement statement = connection.createStatement()) {
+
+ boolean hasResultSet =
+ statement.execute("select avg(s1) from root.sg1.d1 where time>=16
and time<=20");
+ Assert.assertTrue(hasResultSet);
+ try (ResultSet resultSet = statement.getResultSet()) {
+ ResultSetMetaData resultSetMetaData = resultSet.getMetaData();
+ Map<String, Integer> map = new HashMap<>();
+ for (int i = 1; i <= resultSetMetaData.getColumnCount(); i++) {
+ map.put(resultSetMetaData.getColumnName(i), i);
+ }
+ assertEquals(columnNames.length, resultSetMetaData.getColumnCount());
+ int cnt = 0;
+ while (resultSet.next()) {
+ StringBuilder builder = new StringBuilder();
+ // No need to add time column for aggregation query
+ for (String columnName : columnNames) {
+ int index = map.get(columnName);
+ if (builder.length() != 0) {
+ builder.append(",");
+ }
+ builder.append(resultSet.getString(index));
+ }
+ assertEquals(retArray[cnt], builder.toString());
+ cnt++;
+ }
+ assertEquals(1, cnt);
+ }
+ } catch (SQLException e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+}
diff --git
a/tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/BatchData.java
b/tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/BatchData.java
index 5907cfc..981a4f5 100644
--- a/tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/BatchData.java
+++ b/tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/BatchData.java
@@ -870,5 +870,21 @@ public class BatchData {
TsPrimitiveType v = getVector()[subIndex];
return v == null ? null : v.getValue();
}
+
+ @Override
+ public int totalLength() {
+ // aligned timeseries' BatchData length() may return the length of time
column
+ // we need traverse to VectorBatchDataIterator calculate the actual
value column's length
+ int cnt = 0;
+ int readCurArrayIndexSave = BatchData.this.readCurArrayIndex;
+ int readCurListIndexSave = BatchData.this.readCurListIndex;
+ while (hasNext()) {
+ cnt++;
+ next();
+ }
+ BatchData.this.readCurArrayIndex = readCurArrayIndexSave;
+ BatchData.this.readCurListIndex = readCurListIndexSave;
+ return cnt;
+ }
}
}
