You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ja...@apache.org on 2022/05/20 06:06:58 UTC
[iotdb] branch rel/0.13 updated: [ISSUE-5964] Fix bug of aligned timeseries time duplicated aggregation (#5966)
This is an automated email from the ASF dual-hosted git repository.
jackietien pushed a commit to branch rel/0.13
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/rel/0.13 by this push:
new d23156ee34 [ISSUE-5964] Fix bug of aligned timeseries time duplicated aggregation (#5966)
d23156ee34 is described below
commit d23156ee347a860f77d8aeb0fefd9e51614413fd
Author: Haonan <hh...@outlook.com>
AuthorDate: Fri May 20 14:06:53 2022 +0800
[ISSUE-5964] Fix bug of aligned timeseries time duplicated aggregation (#5966)
---
.../IoTDBAggregationWithoutValueFilterIT.java | 44 ++++++++++++++++++++++
.../querycontext/AlignedReadOnlyMemChunk.java | 42 ++++++++++++++++++++-
2 files changed, 84 insertions(+), 2 deletions(-)
diff --git a/integration/src/test/java/org/apache/iotdb/db/integration/aligned/IoTDBAggregationWithoutValueFilterIT.java b/integration/src/test/java/org/apache/iotdb/db/integration/aligned/IoTDBAggregationWithoutValueFilterIT.java
index 45333dce1d..4769083250 100644
--- a/integration/src/test/java/org/apache/iotdb/db/integration/aligned/IoTDBAggregationWithoutValueFilterIT.java
+++ b/integration/src/test/java/org/apache/iotdb/db/integration/aligned/IoTDBAggregationWithoutValueFilterIT.java
@@ -453,4 +453,48 @@ public class IoTDBAggregationWithoutValueFilterIT {
fail(e.getMessage());
}
}
+
+ @Test
+ public void sumAlignedWithDuplicateTimeTest() throws ClassNotFoundException {
+ double[][] retArray = {{33.0, 33.0}};
+ String[] columnNames = {"sum(root.sg5.d1.success)", "sum(root.sg5.d1.fail)"};
+ 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()) {
+ statement.execute("insert into root.sg5.d1(timestamp,success,fail) aligned values(1,10,10);");
+ statement.execute("insert into root.sg5.d1(timestamp,success,fail) aligned values(1,11,11);");
+ statement.execute("insert into root.sg5.d1(timestamp,success,fail) aligned values(2,11,11);");
+ statement.execute("insert into root.sg5.d1(timestamp,success,fail) aligned values(3,11,11);");
+ boolean hasResultSet = statement.execute("select sum(success), sum(fail) from root.sg5.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];
+ 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/main/java/org/apache/iotdb/db/engine/querycontext/AlignedReadOnlyMemChunk.java b/server/src/main/java/org/apache/iotdb/db/engine/querycontext/AlignedReadOnlyMemChunk.java
index f0d7cb1be5..781311a56a 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/querycontext/AlignedReadOnlyMemChunk.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/querycontext/AlignedReadOnlyMemChunk.java
@@ -32,6 +32,7 @@ import org.apache.iotdb.tsfile.file.metadata.enums.TSEncoding;
import org.apache.iotdb.tsfile.file.metadata.statistics.Statistics;
import org.apache.iotdb.tsfile.read.common.TimeRange;
import org.apache.iotdb.tsfile.read.reader.IPointReader;
+import org.apache.iotdb.tsfile.utils.Pair;
import org.apache.iotdb.tsfile.write.schema.IMeasurementSchema;
import org.apache.iotdb.tsfile.write.schema.VectorMeasurementSchema;
@@ -95,12 +96,26 @@ public class AlignedReadOnlyMemChunk extends ReadOnlyMemChunk {
new ChunkMetadata(measurementUid, TSDataType.VECTOR, 0, timeStatistics);
List<IChunkMetadata> valueChunkMetadataList = new ArrayList<>();
// update time chunk
+ boolean[] timeDuplicateInfo = null;
for (int row = 0; row < alignedChunkData.rowCount(); row++) {
- timeStatistics.update(alignedChunkData.getTime(row));
+ if (row == alignedChunkData.rowCount() - 1
+ || alignedChunkData.getTime(row) != alignedChunkData.getTime(row + 1)) {
+ timeStatistics.update(alignedChunkData.getTime(row));
+ } else {
+ if (timeDuplicateInfo == null) {
+ timeDuplicateInfo = new boolean[alignedChunkData.rowCount()];
+ }
+ timeDuplicateInfo[row] = true;
+ }
}
timeStatistics.setEmpty(false);
// update value chunk
for (int column = 0; column < measurementList.size(); column++) {
+ // Pair of Time and Index
+ Pair<Long, Integer> lastValidPointIndexForTimeDupCheck = null;
+ if (timeDuplicateInfo != null) {
+ lastValidPointIndexForTimeDupCheck = new Pair<>(Long.MIN_VALUE, null);
+ }
Statistics valueStatistics = Statistics.getStatsByType(dataTypeList.get(column));
IChunkMetadata valueChunkMetadata =
new ChunkMetadata(
@@ -112,7 +127,30 @@ public class AlignedReadOnlyMemChunk extends ReadOnlyMemChunk {
}
for (int row = 0; row < alignedChunkData.rowCount(); row++) {
long time = alignedChunkData.getTime(row);
- int originRowIndex = alignedChunkData.getValueIndex(row);
+ // skip time duplicated rows
+ if (timeDuplicateInfo != null) {
+ if (!alignedChunkData.isValueMarked(alignedChunkData.getValueIndex(row), column)) {
+ lastValidPointIndexForTimeDupCheck.left = time;
+ lastValidPointIndexForTimeDupCheck.right = alignedChunkData.getValueIndex(row);
+ }
+ if (timeDuplicateInfo[row]) {
+ continue;
+ }
+ }
+ // The part of code solves the following problem:
+ // Time: 1,2,2,3
+ // Value: 1,2,null,null
+ // When rowIndex:1, pair(min,null), timeDuplicateInfo:false, write(T:1,V:1)
+ // When rowIndex:2, pair(2,2), timeDuplicateInfo:true, skip writing value
+ // When rowIndex:3, pair(2,2), timeDuplicateInfo:false, T:2!=air.left:2, write(T:2,V:2)
+ // When rowIndex:4, pair(2,2), timeDuplicateInfo:false, T:3!=pair.left:2, write(T:3,V:null)
+ int originRowIndex;
+ if (lastValidPointIndexForTimeDupCheck != null
+ && (alignedChunkData.getTime(row) == lastValidPointIndexForTimeDupCheck.left)) {
+ originRowIndex = lastValidPointIndexForTimeDupCheck.right;
+ } else {
+ originRowIndex = alignedChunkData.getValueIndex(row);
+ }
boolean isNull = alignedChunkData.isValueMarked(originRowIndex, column);
if (isNull) {
continue;