You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by le...@apache.org on 2024/02/05 08:26:26 UTC
(iotdb) 06/30: add iteration
This is an automated email from the ASF dual-hosted git repository.
leirui pushed a commit to branch research/LTS-visualization
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 5b4209916ce0495c2b5650c00e8e8713a4702fc1
Author: Lei Rui <10...@qq.com>
AuthorDate: Sun Jan 28 15:23:46 2024 +0800
add iteration
---
.../groupby/GroupByWithoutValueFilterDataSet.java | 12 +-
.../groupby/LocalGroupByExecutorTri_ILTS.java | 316 +++++++++++++++++++++
.../groupby/LocalGroupByExecutorTri_LTTB.java | 151 +---------
.../iotdb/db/integration/tri/MyTest_ILTS.java | 248 ++++++++++++++++
4 files changed, 569 insertions(+), 158 deletions(-)
diff --git a/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/GroupByWithoutValueFilterDataSet.java b/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/GroupByWithoutValueFilterDataSet.java
index 913158a5021..c349d9dc3c9 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/GroupByWithoutValueFilterDataSet.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/GroupByWithoutValueFilterDataSet.java
@@ -142,14 +142,9 @@ public class GroupByWithoutValueFilterDataSet extends GroupByEngineDataSet {
}
// } else if (CONFIG.getEnableTri().equals("M4LTTB")) {
// // TODO
- else if (CONFIG.getEnableTri().equals("LTTB")) {
- // TODO
+ else if (CONFIG.getEnableTri().equals("LTTB") || CONFIG.getEnableTri().equals("ILTS")) {
return nextWithoutConstraintTri_LTTB();
- }
- // } else if (CONFIG.getEnableTri().equals("ILTS")) {
- // // TODO
- // }
- else {
+ } else {
return nextWithoutConstraint_raw();
}
}
@@ -494,12 +489,11 @@ public class GroupByWithoutValueFilterDataSet extends GroupByEngineDataSet {
return new LocalGroupByExecutor(
path, allSensors, dataType, context, timeFilter, fileFilter, ascending);
} else if (CONFIG.getEnableTri().equals("LTTB")) {
- // TODO
return new LocalGroupByExecutorTri_LTTB(
path, allSensors, dataType, context, timeFilter, fileFilter, ascending);
} else if (CONFIG.getEnableTri().equals("ILTS")) {
// TODO
- return new LocalGroupByExecutor(
+ return new LocalGroupByExecutorTri_ILTS(
path, allSensors, dataType, context, timeFilter, fileFilter, ascending);
}
// deprecated below
diff --git a/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/LocalGroupByExecutorTri_ILTS.java b/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/LocalGroupByExecutorTri_ILTS.java
new file mode 100644
index 00000000000..73f9fd395d7
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/LocalGroupByExecutorTri_ILTS.java
@@ -0,0 +1,316 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.query.dataset.groupby;
+
+import org.apache.iotdb.db.conf.IoTDBConfig;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.engine.querycontext.QueryDataSource;
+import org.apache.iotdb.db.exception.StorageEngineException;
+import org.apache.iotdb.db.exception.query.QueryProcessException;
+import org.apache.iotdb.db.metadata.PartialPath;
+import org.apache.iotdb.db.query.aggregation.AggregateResult;
+import org.apache.iotdb.db.query.aggregation.impl.MinValueAggrResult;
+import org.apache.iotdb.db.query.context.QueryContext;
+import org.apache.iotdb.db.query.control.QueryResourceManager;
+import org.apache.iotdb.db.query.filter.TsFileFilter;
+import org.apache.iotdb.db.query.reader.series.SeriesReader;
+import org.apache.iotdb.db.utils.FileLoaderUtils;
+import org.apache.iotdb.tsfile.exception.write.UnSupportedDataTypeException;
+import org.apache.iotdb.tsfile.file.metadata.ChunkMetadata;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.file.metadata.statistics.MinMaxInfo;
+import org.apache.iotdb.tsfile.read.common.ChunkSuit4Tri;
+import org.apache.iotdb.tsfile.read.common.IOMonitor2;
+import org.apache.iotdb.tsfile.read.filter.GroupByFilter;
+import org.apache.iotdb.tsfile.read.filter.basic.Filter;
+import org.apache.iotdb.tsfile.read.reader.page.PageReader;
+import org.apache.iotdb.tsfile.utils.Pair;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+public class LocalGroupByExecutorTri_ILTS implements GroupByExecutor {
+
+ private static final IoTDBConfig CONFIG = IoTDBDescriptor.getInstance().getConfig();
+ private static final Logger M4_CHUNK_METADATA = LoggerFactory.getLogger("M4_CHUNK_METADATA");
+
+ // Aggregate result buffer of this path
+ private final List<AggregateResult> results = new ArrayList<>();
+
+ // private List<ChunkSuit4Tri> currentChunkList;
+
+ // private final List<ChunkSuit4Tri> futureChunkList = new ArrayList<>();
+
+ // keys: 0,1,...,(int) Math.floor((endTime * 1.0 - startTime) / interval)-1
+ private final Map<Integer, List<ChunkSuit4Tri>> splitChunkList = new HashMap<>();
+
+ private final long p1t = CONFIG.getP1t();
+ private final double p1v = CONFIG.getP1v();
+ private final long pnt = CONFIG.getPnt();
+ private final double pnv = CONFIG.getPnv();
+
+ private long lt = p1t;
+ private double lv = p1v;
+
+ private final int N1; // 分桶数
+
+ private static final int numIterations = 8;
+
+ private Filter timeFilter;
+
+ public LocalGroupByExecutorTri_ILTS(
+ PartialPath path,
+ Set<String> allSensors,
+ TSDataType dataType,
+ QueryContext context,
+ Filter timeFilter,
+ TsFileFilter fileFilter,
+ boolean ascending)
+ throws StorageEngineException, QueryProcessException {
+ // long start = System.nanoTime();
+
+ // get all data sources
+ QueryDataSource queryDataSource =
+ QueryResourceManager.getInstance().getQueryDataSource(path, context, this.timeFilter);
+
+ // update filter by TTL
+ this.timeFilter = queryDataSource.updateFilterUsingTTL(timeFilter);
+
+ SeriesReader seriesReader =
+ new SeriesReader(
+ path,
+ allSensors,
+ // fix bug: here use the aggregation type as the series data type,
+ // not using pageReader.getAllSatisfiedPageData is ok
+ dataType,
+ context,
+ queryDataSource,
+ timeFilter,
+ null,
+ fileFilter,
+ ascending);
+
+ // unpackAllOverlappedFilesToTimeSeriesMetadata
+ try {
+ // : this might be bad to load all chunk metadata at first
+ List<ChunkSuit4Tri> futureChunkList =
+ new ArrayList<>(seriesReader.getAllChunkMetadatas4Tri()); // no need sort here
+ // arrange futureChunkList into each bucket
+ GroupByFilter groupByFilter = (GroupByFilter) timeFilter;
+ long startTime = groupByFilter.getStartTime();
+ long endTime = groupByFilter.getEndTime();
+ long interval = groupByFilter.getInterval();
+ N1 = (int) Math.floor((endTime * 1.0 - startTime) / interval); // 分桶数
+ for (ChunkSuit4Tri chunkSuit4Tri : futureChunkList) {
+ ChunkMetadata chunkMetadata = chunkSuit4Tri.chunkMetadata;
+ long chunkMinTime = chunkMetadata.getStartTime();
+ long chunkMaxTime = chunkMetadata.getEndTime();
+ int idx1 = (int) Math.floor((chunkMinTime - startTime) * 1.0 / interval);
+ int idx2 = (int) Math.floor((chunkMaxTime - startTime) * 1.0 / interval);
+ idx2 = (int) Math.min(idx2, N1 - 1);
+ for (int i = idx1; i <= idx2; i++) {
+ splitChunkList.computeIfAbsent(i, k -> new ArrayList<>());
+ splitChunkList.get(i).add(chunkSuit4Tri);
+ }
+ }
+
+ } catch (IOException e) {
+ throw new QueryProcessException(e.getMessage());
+ }
+
+ // IOMonitor2.addMeasure(Operation.M4_LSM_INIT_LOAD_ALL_CHUNKMETADATAS, System.nanoTime() -
+ // start);
+ }
+
+ @Override
+ public void addAggregateResult(AggregateResult aggrResult) {
+ results.add(aggrResult);
+ }
+
+ @Override
+ public List<AggregateResult> calcResult(
+ long curStartTime, long curEndTime, long startTime, long endTime, long interval)
+ throws IOException {
+ // 这里用calcResult一次返回所有buckets结果（可以把MinValueAggrResult的value设为string类型，
+ // 那就把所有buckets结果作为一个string返回。这样的话返回的[t]是没有意义的，只取valueString）
+ // 而不是像MinMax那样在nextWithoutConstraintTri_MinMax()里调用calcResult每次计算一个bucket
+ StringBuilder series_final = new StringBuilder();
+
+ // clear result cache
+ for (AggregateResult result : results) {
+ result.reset();
+ }
+
+ long[] lastIter_t = new long[N1]; // N1不包括全局首尾点
+ double[] lastIter_v = new double[N1]; // N1不包括全局首尾点
+ for (int num = 0; num < numIterations; num++) {
+ StringBuilder series = new StringBuilder();
+ // 全局首点
+ series.append(p1v).append("[").append(p1t).append("]").append(",");
+ // 遍历分桶 Assume no empty buckets
+ for (int b = 0; b < N1; b++) {
+ long rt = 0; // must initialize as zero, because may be used as sum for average
+ double rv = 0; // must initialize as zero, because may be used as sum for average
+ // 计算右边桶的固定点
+ if (b == N1 - 1) { // 最后一个桶
+ // 全局尾点
+ rt = pnt;
+ rv = pnv;
+ } else { // 不是最后一个桶
+ if (num == 0) { // 是第一次迭代的话，就使用右边桶的平均点
+ // ========计算右边桶的平均点========
+ List<ChunkSuit4Tri> chunkSuit4TriList = splitChunkList.get(b + 1);
+ long rightStartTime = startTime + (b + 1) * interval;
+ long rightEndTime = startTime + (b + 2) * interval;
+ int cnt = 0;
+ // 遍历所有与右边桶overlap的chunks
+ for (ChunkSuit4Tri chunkSuit4Tri : chunkSuit4TriList) {
+ TSDataType dataType = chunkSuit4Tri.chunkMetadata.getDataType();
+ if (dataType != TSDataType.DOUBLE) {
+ throw new UnSupportedDataTypeException(String.valueOf(dataType));
+ }
+ // 1. load page data if it hasn't been loaded
+ if (chunkSuit4Tri.pageReader == null) {
+ chunkSuit4Tri.pageReader =
+ FileLoaderUtils.loadPageReaderList4CPV(
+ chunkSuit4Tri.chunkMetadata, this.timeFilter);
+ // ATTENTION: YOU HAVE TO ENSURE THAT THERE IS ONLY ONE PAGE IN A CHUNK,
+ // BECAUSE THE WHOLE IMPLEMENTATION IS BASED ON THIS ASSUMPTION.
+ // OTHERWISE, PAGEREADER IS FOR THE FIRST PAGE IN THE CHUNK WHILE
+ // STEPREGRESS IS FOR THE LAST PAGE IN THE CHUNK (THE MERGE OF STEPREGRESS IS
+ // ASSIGN DIRECTLY), WHICH WILL INTRODUCE BUGS!
+ }
+ // 2. 计算平均点
+ PageReader pageReader = chunkSuit4Tri.pageReader;
+ for (int j = 0; j < chunkSuit4Tri.chunkMetadata.getStatistics().getCount(); j++) {
+ long timestamp = pageReader.timeBuffer.getLong(j * 8);
+ if (timestamp < rightStartTime) {
+ continue;
+ } else if (timestamp >= rightEndTime) {
+ break;
+ } else { // rightStartTime<=t<rightEndTime
+ ByteBuffer valueBuffer = pageReader.valueBuffer;
+ double v = valueBuffer.getDouble(pageReader.timeBufferLength + j * 8);
+ rt += timestamp;
+ rv += v;
+ cnt++;
+ }
+ }
+ }
+ rt = rt / cnt;
+ rv = rv / cnt;
+ } else { // 不是第一次迭代也不是最后一个桶的话，就使用上一轮迭代右边桶的采样点
+ rt = lastIter_t[b + 1];
+ rv = lastIter_v[b + 1];
+ }
+ }
+ // ========找到当前桶内距离lr连线最远的点========
+ double maxArea = -1;
+ long select_t = -1;
+ double select_v = -1;
+ List<ChunkSuit4Tri> chunkSuit4TriList = splitChunkList.get(b);
+ long localCurStartTime = startTime + (b) * interval;
+ long localCurEndTime = startTime + (b + 1) * interval;
+ // 遍历所有与当前桶overlap的chunks
+ for (ChunkSuit4Tri chunkSuit4Tri : chunkSuit4TriList) {
+ TSDataType dataType = chunkSuit4Tri.chunkMetadata.getDataType();
+ if (dataType != TSDataType.DOUBLE) {
+ throw new UnSupportedDataTypeException(String.valueOf(dataType));
+ }
+ // load page data if it hasn't been loaded
+ if (chunkSuit4Tri.pageReader == null) {
+ chunkSuit4Tri.pageReader =
+ FileLoaderUtils.loadPageReaderList4CPV(
+ chunkSuit4Tri.chunkMetadata, this.timeFilter);
+ // ATTENTION: YOU HAVE TO ENSURE THAT THERE IS ONLY ONE PAGE IN A CHUNK,
+ // BECAUSE THE WHOLE IMPLEMENTATION IS BASED ON THIS ASSUMPTION.
+ // OTHERWISE, PAGEREADER IS FOR THE FIRST PAGE IN THE CHUNK WHILE
+ // STEPREGRESS IS FOR THE LAST PAGE IN THE CHUNK (THE MERGE OF STEPREGRESS IS
+ // ASSIGN DIRECTLY), WHICH WILL INTRODUCE BUGS!
+ }
+ PageReader pageReader = chunkSuit4Tri.pageReader;
+ for (int j = 0; j < chunkSuit4Tri.chunkMetadata.getStatistics().getCount(); j++) {
+ long timestamp = pageReader.timeBuffer.getLong(j * 8);
+ if (timestamp < localCurStartTime) {
+ continue;
+ } else if (timestamp >= localCurEndTime) {
+ break;
+ } else { // localCurStartTime<=t<localCurEndTime
+ ByteBuffer valueBuffer = pageReader.valueBuffer;
+ double v = valueBuffer.getDouble(pageReader.timeBufferLength + j * 8);
+ double area = IOMonitor2.calculateTri(lt, lv, timestamp, v, rt, rv);
+ if (area > maxArea) {
+ maxArea = area;
+ select_t = timestamp;
+ select_v = v;
+ }
+ }
+ }
+ }
+ // 记录结果
+ series.append(select_v).append("[").append(select_t).append("]").append(",");
+
+ // 更新lt,lv
+ // 下一个桶自然地以select_t, select_v作为左桶固定点
+ lt = select_t;
+ lv = select_v;
+ // 记录本轮迭代本桶选点
+ lastIter_t[b] = select_t;
+ lastIter_v[b] = select_v;
+ } // 遍历分桶结束
+ // 全局尾点
+ series.append(pnv).append("[").append(pnt).append("]").append(",");
+
+ System.out.println(series);
+ } // end Iterations
+
+ // 全局首点
+ series_final.append(p1v).append("[").append(p1t).append("]").append(",");
+ for (int i = 0; i < lastIter_t.length; i++) {
+ series_final.append(lastIter_v[i]).append("[").append(lastIter_t[i]).append("]").append(",");
+ }
+ // 全局尾点
+ series_final.append(pnv).append("[").append(pnt).append("]").append(",");
+ MinValueAggrResult minValueAggrResult = (MinValueAggrResult) results.get(0);
+ minValueAggrResult.updateResult(new MinMaxInfo<>(series_final.toString(), 0));
+
+ return results;
+ }
+
+ @Override
+ public Pair<Long, Object> peekNextNotNullValue(long nextStartTime, long nextEndTime)
+ throws IOException {
+ throw new IOException("no implemented");
+ }
+
+ @Override
+ public List<AggregateResult> calcResult(long curStartTime, long curEndTime)
+ throws IOException, QueryProcessException {
+ throw new IOException("no implemented");
+ }
+}
diff --git a/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/LocalGroupByExecutorTri_LTTB.java b/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/LocalGroupByExecutorTri_LTTB.java
index 9ec058e2059..be84be70c38 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/LocalGroupByExecutorTri_LTTB.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/LocalGroupByExecutorTri_LTTB.java
@@ -26,7 +26,6 @@ import org.apache.iotdb.db.exception.StorageEngineException;
import org.apache.iotdb.db.exception.query.QueryProcessException;
import org.apache.iotdb.db.metadata.PartialPath;
import org.apache.iotdb.db.query.aggregation.AggregateResult;
-import org.apache.iotdb.db.query.aggregation.impl.MaxValueAggrResult;
import org.apache.iotdb.db.query.aggregation.impl.MinValueAggrResult;
import org.apache.iotdb.db.query.context.QueryContext;
import org.apache.iotdb.db.query.control.QueryResourceManager;
@@ -37,7 +36,6 @@ import org.apache.iotdb.tsfile.exception.write.UnSupportedDataTypeException;
import org.apache.iotdb.tsfile.file.metadata.ChunkMetadata;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.file.metadata.statistics.MinMaxInfo;
-import org.apache.iotdb.tsfile.file.metadata.statistics.Statistics;
import org.apache.iotdb.tsfile.read.common.ChunkSuit4Tri;
import org.apache.iotdb.tsfile.read.common.IOMonitor2;
import org.apache.iotdb.tsfile.read.filter.GroupByFilter;
@@ -78,7 +76,6 @@ public class LocalGroupByExecutorTri_LTTB implements GroupByExecutor {
private long lt = p1t;
private double lv = p1v;
- // private boolean[] notEmptyBucket;
private final int N1; // 分桶数
@@ -119,8 +116,8 @@ public class LocalGroupByExecutorTri_LTTB implements GroupByExecutor {
// unpackAllOverlappedFilesToTimeSeriesMetadata
try {
// : this might be bad to load all chunk metadata at first
- List<ChunkSuit4Tri> futureChunkList = new ArrayList<>();
- futureChunkList.addAll(seriesReader.getAllChunkMetadatas4Tri()); // no need sort here
+ List<ChunkSuit4Tri> futureChunkList =
+ new ArrayList<>(seriesReader.getAllChunkMetadatas4Tri()); // no need sort here
// arrange futureChunkList into each bucket
GroupByFilter groupByFilter = (GroupByFilter) timeFilter;
long startTime = groupByFilter.getStartTime();
@@ -153,39 +150,6 @@ public class LocalGroupByExecutorTri_LTTB implements GroupByExecutor {
results.add(aggrResult);
}
- // private void getCurrentChunkListFromFutureChunkList(long curStartTime, long curEndTime) {
- // // IOMonitor2.M4_LSM_status = Operation.M4_LSM_MERGE_M4_TIME_SPAN;
- //
- // // empty currentChunkList
- // currentChunkList = new ArrayList<>();
- //
- // // iterate futureChunkList
- // ListIterator<ChunkSuit4Tri> itr = futureChunkList.listIterator();
- // while (itr.hasNext()) {
- // ChunkSuit4Tri chunkSuit4Tri = itr.next();
- // ChunkMetadata chunkMetadata = chunkSuit4Tri.chunkMetadata;
- // long chunkMinTime = chunkMetadata.getStartTime();
- // long chunkMaxTime = chunkMetadata.getEndTime();
- // if (chunkMaxTime < curStartTime) {
- // // the chunk falls on the left side of the current M4 interval Ii
- // itr.remove();
- // } else if (chunkMinTime >= curEndTime) {
- // // the chunk falls on the right side of the current M4 interval Ii,
- // // and since futureChunkList is ordered by the startTime of chunkMetadata,
- // // the loop can be terminated early.
- // break;
- // } else if (chunkMaxTime < curEndTime) {
- // // this chunk is not related to buckets later
- // currentChunkList.add(chunkSuit4Tri);
- // itr.remove();
- // } else {
- // // this chunk is overlapped with the right border of the current bucket
- // currentChunkList.add(chunkSuit4Tri);
- // // still keep it in the futureChunkList
- // }
- // }
- // }
-
@Override
public List<AggregateResult> calcResult(
long curStartTime, long curEndTime, long startTime, long endTime, long interval)
@@ -313,120 +277,9 @@ public class LocalGroupByExecutorTri_LTTB implements GroupByExecutor {
MinValueAggrResult minValueAggrResult = (MinValueAggrResult) results.get(0);
minValueAggrResult.updateResult(new MinMaxInfo<>(series.toString(), 0));
- // int idx = (int) Math.floor((curStartTime - startTime) * 1.0 / interval);
- // System.out.println("idx=" + idx);
- // List<ChunkSuit4Tri> chunkSuit4TriList = splitChunkList.get(idx);
- // if (chunkSuit4TriList != null) {
- // for (ChunkSuit4Tri chunkSuit4Tri : chunkSuit4TriList) {
- // System.out.println(chunkSuit4Tri.chunkMetadata.getStartTime());
- // }
- // }
-
- // long start = System.nanoTime();
- // getCurrentChunkListFromFutureChunkList(curStartTime, curEndTime);
- // IOMonitor2.addMeasure(Operation.M4_LSM_MERGE_M4_TIME_SPAN, System.nanoTime() - start);
-
- // if (currentChunkList.size() == 0) {
- // return results;
- // }
-
- // start = System.nanoTime();
- // calculateMinMax(currentChunkList, curStartTime, curEndTime);
- // IOMonitor2.addMeasure(Operation.M4_LSM_FP, System.nanoTime() - start);
-
return results;
}
- private void calculateMinMax(
- List<ChunkSuit4Tri> currentChunkList, long curStartTime, long curEndTime) throws IOException {
- for (ChunkSuit4Tri chunkSuit4Tri : currentChunkList) {
-
- Statistics statistics = chunkSuit4Tri.chunkMetadata.getStatistics();
-
- if (canUseStatistics(chunkSuit4Tri, curStartTime, curEndTime)) {
- // update BP
- MinValueAggrResult minValueAggrResult = (MinValueAggrResult) results.get(0);
- minValueAggrResult.updateResult(
- new MinMaxInfo<>(statistics.getMinValue(), statistics.getBottomTimestamp()));
- // update TP
- MaxValueAggrResult maxValueAggrResult = (MaxValueAggrResult) results.get(1);
- maxValueAggrResult.updateResult(
- new MinMaxInfo<>(statistics.getMaxValue(), statistics.getTopTimestamp()));
- } else { // cannot use statistics directly
-
- Comparable<Object> minVal = null;
- long bottomTime = -1;
- Comparable<Object> maxVal = null;
- long topTime = -1;
-
- // 1. load page data if it hasn't been loaded
- if (chunkSuit4Tri.pageReader == null) {
- chunkSuit4Tri.pageReader =
- FileLoaderUtils.loadPageReaderList4CPV(chunkSuit4Tri.chunkMetadata, this.timeFilter);
- // ATTENTION: YOU HAVE TO ENSURE THAT THERE IS ONLY ONE PAGE IN A CHUNK,
- // BECAUSE THE WHOLE IMPLEMENTATION IS BASED ON THIS ASSUMPTION.
- // OTHERWISE, PAGEREADER IS FOR THE FIRST PAGE IN THE CHUNK WHILE
- // STEPREGRESS IS FOR THE LAST PAGE IN THE CHUNK (THE MERGE OF STEPREGRESS IS
- // ASSIGN DIRECTLY), WHICH WILL INTRODUCE BUGS!
- }
-
- int count = chunkSuit4Tri.chunkMetadata.getStatistics().getCount();
- PageReader pageReader = chunkSuit4Tri.pageReader;
- for (int i = chunkSuit4Tri.lastReadPos; i < count; i++) {
- long timestamp = pageReader.timeBuffer.getLong(i * 8);
- if (timestamp < curStartTime) {
- // 2. read from lastReadPos until the first point fallen within this bucket (if it
- // exists)
- continue;
- } else if (timestamp >= curEndTime) {
- // 3. traverse until the first point fallen right this bucket, also remember to update
- // lastReadPos
- chunkSuit4Tri.lastReadPos = i;
- break;
- } else {
- // 4. update MinMax by traversing points fallen within this bucket
- ByteBuffer valueBuffer = pageReader.valueBuffer;
- TSDataType dataType = chunkSuit4Tri.chunkMetadata.getDataType();
- Object v;
- switch (dataType) {
- // case INT64:
- // v = valueBuffer.getLong(pageReader.timeBufferLength + i * 8);
- // break;
- case DOUBLE:
- v = valueBuffer.getDouble(pageReader.timeBufferLength + i * 8);
- break;
- default:
- throw new UnSupportedDataTypeException(String.valueOf(dataType));
- }
- if (minVal == null || minVal.compareTo(v) > 0) {
- minVal = (Comparable<Object>) v;
- bottomTime = timestamp;
- }
- if (maxVal == null || maxVal.compareTo(v) < 0) {
- maxVal = (Comparable<Object>) v;
- topTime = timestamp;
- }
- }
- }
- // 4. update MinMax by traversing points fallen within this bucket
- if (minVal != null) {
- // update BP
- MinValueAggrResult minValueAggrResult = (MinValueAggrResult) results.get(0);
- minValueAggrResult.updateResult(new MinMaxInfo<>(minVal, bottomTime));
- // update TP
- MaxValueAggrResult maxValueAggrResult = (MaxValueAggrResult) results.get(1);
- maxValueAggrResult.updateResult(new MinMaxInfo<>(maxVal, topTime));
- }
- }
- }
- }
-
- public boolean canUseStatistics(ChunkSuit4Tri chunkSuit4Tri, long curStartTime, long curEndTime) {
- long TP_t = chunkSuit4Tri.chunkMetadata.getStatistics().getTopTimestamp();
- long BP_t = chunkSuit4Tri.chunkMetadata.getStatistics().getBottomTimestamp();
- return TP_t >= curStartTime && TP_t < curEndTime && BP_t >= curStartTime && BP_t < curEndTime;
- }
-
@Override
public Pair<Long, Object> peekNextNotNullValue(long nextStartTime, long nextEndTime)
throws IOException {
diff --git a/server/src/test/java/org/apache/iotdb/db/integration/tri/MyTest_ILTS.java b/server/src/test/java/org/apache/iotdb/db/integration/tri/MyTest_ILTS.java
new file mode 100644
index 00000000000..d2381e6fa75
--- /dev/null
+++ b/server/src/test/java/org/apache/iotdb/db/integration/tri/MyTest_ILTS.java
@@ -0,0 +1,248 @@
+/*
+ * 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.tri;
+
+import org.apache.iotdb.db.conf.IoTDBConfig;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.engine.compaction.CompactionStrategy;
+import org.apache.iotdb.db.utils.EnvironmentUtils;
+import org.apache.iotdb.jdbc.Config;
+import org.apache.iotdb.tsfile.common.conf.TSFileDescriptor;
+
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.ResultSet;
+import java.sql.Statement;
+import java.util.Locale;
+
+import static org.junit.Assert.fail;
+
+public class MyTest_ILTS {
+
+ /*
+ * Sql format: SELECT min_value(s0), max_value(s0) ROM root.xx group by ([tqs,tqe),IntervalLength).
+ * Requirements:
+ * (1) Don't change the sequence of the above two aggregates
+ * (2) Assume each chunk has only one page.
+ * (3) Assume all chunks are sequential and no deletes.
+ * (4) Assume plain encoding, UNCOMPRESSED, Long or Double data type, no compaction
+ * (5) Assume no empty bucket
+ */
+ private static final String TIMESTAMP_STR = "Time";
+
+ private static String[] creationSqls =
+ new String[] {
+ "SET STORAGE GROUP TO root.vehicle.d0",
+ "CREATE TIMESERIES root.vehicle.d0.s0 WITH DATATYPE=DOUBLE, ENCODING=PLAIN",
+ // IoTDB int data type does not support plain encoding, so use long data type
+ };
+
+ private final String d0s0 = "root.vehicle.d0.s0";
+
+ private static final String insertTemplate =
+ "INSERT INTO root.vehicle.d0(timestamp,s0)" + " VALUES(%d,%f)";
+
+ private static final IoTDBConfig config = IoTDBDescriptor.getInstance().getConfig();
+
+ @Before
+ public void setUp() throws Exception {
+ TSFileDescriptor.getInstance().getConfig().setTimeEncoder("PLAIN");
+ config.setTimestampPrecision("ms");
+ config.setCompactionStrategy(CompactionStrategy.NO_COMPACTION);
+
+ config.setEnableTri("ILTS");
+
+ config.setEnableCPV(false);
+ TSFileDescriptor.getInstance().getConfig().setEnableMinMaxLSM(false);
+ TSFileDescriptor.getInstance().getConfig().setUseStatistics(false);
+
+ EnvironmentUtils.envSetUp();
+ Class.forName(Config.JDBC_DRIVER_NAME);
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ EnvironmentUtils.cleanEnv();
+ }
+
+ @Test
+ public void test3_2() {
+ prepareData3_2();
+
+ String res = "5.0[1],10.0[2],2.0[40],5.0[55],20.0[62],1.0[90],7.0[102],";
+ try (Connection connection =
+ DriverManager.getConnection("jdbc:iotdb://127.0.0.1:6667/", "root", "root");
+ Statement statement = connection.createStatement()) {
+ boolean hasResultSet =
+ statement.execute(
+ "SELECT min_value(s0)"
+ // TODO not real min_value here, actually controlled by enableTri
+ + " FROM root.vehicle.d0 group by ([2,102),20ms)");
+ // (102-2)/(7-2)=20ms
+ // note keep no empty buckets
+
+ Assert.assertTrue(hasResultSet);
+ try (ResultSet resultSet = statement.getResultSet()) {
+ int i = 0;
+ while (resultSet.next()) {
+ String ans = resultSet.getString(2);
+ // for LTTB all results are in the value string of MinValueAggrResult
+ // 因此对于LTTB来说，MinValueAggrResult的[t]也无意义
+ ans = ans.substring(0, ans.length() - 3);
+ System.out.println(ans);
+ Assert.assertEquals(res, ans);
+ }
+ }
+ } catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ private static void prepareData3_2() {
+ // data:
+ // https://user-images.githubusercontent.com/33376433/152003603-6b4e7494-00ff-47e4-bf6e-cab3c8600ce2.png
+ // slightly modified
+ try (Connection connection =
+ DriverManager.getConnection(
+ Config.IOTDB_URL_PREFIX + "127.0.0.1:6667/", "root", "root");
+ Statement statement = connection.createStatement()) {
+
+ for (String sql : creationSqls) {
+ statement.execute(sql);
+ }
+
+ long[] t = new long[] {1, 2, 10, 20, 22, 30, 40, 55, 60, 62, 65, 70, 72, 80, 90, 102};
+ double[] v = new double[] {5, 10, 1, 5, 4, 8, 2, 5, 15, 20, 8, 18, 4, 11, 1, 7};
+ config.setP1t(t[0]);
+ config.setP1v(v[0]);
+ config.setPnt(t[t.length - 1]);
+ config.setPnv(v[v.length - 1]);
+
+ for (int i = 0; i < t.length; i++) {
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, t[i], v[i]));
+ if ((i + 1) % 4 == 0) {
+ statement.execute("FLUSH");
+ }
+ }
+ statement.execute("FLUSH");
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+ }
+
+ @Test
+ public void test3() {
+ prepareData3();
+
+ String res =
+ "-1.2079272[0],1.101946[200],-0.523204[300],0.145359[500],-1.014322[700],"
+ + "0.532565[900],-0.122525[1200],-0.676077[1300],0.809559[1500],0.315869[1800],"
+ + "-0.413534[1900],-0.0211206[2100],";
+ try (Connection connection =
+ DriverManager.getConnection("jdbc:iotdb://127.0.0.1:6667/", "root", "root");
+ Statement statement = connection.createStatement()) {
+ boolean hasResultSet =
+ statement.execute(
+ "SELECT min_value(s0)"
+ // TODO not real min_value here, actually controlled by enableTri
+ + " FROM root.vehicle.d0 group by ([100,2100),200ms)");
+ // (tn-t2)/(nout-2)=(2100-100)/(12-2)=2000/10=200
+
+ Assert.assertTrue(hasResultSet);
+ try (ResultSet resultSet = statement.getResultSet()) {
+ int i = 0;
+ while (resultSet.next()) {
+ String ans = resultSet.getString(2);
+ // for LTTB all results are in the value string of MinValueAggrResult
+ // 因此对于LTTB来说，MinValueAggrResult的[t]也无意义
+ ans = ans.substring(0, ans.length() - 3);
+ System.out.println(ans);
+ Assert.assertEquals(res, ans);
+ }
+ }
+ } catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ private static void prepareData3() {
+ try (Connection connection =
+ DriverManager.getConnection(
+ Config.IOTDB_URL_PREFIX + "127.0.0.1:6667/", "root", "root");
+ Statement statement = connection.createStatement()) {
+
+ for (String sql : creationSqls) {
+ statement.execute(sql);
+ }
+
+ int[] t =
+ new int[] {
+ 0, 100, 200, 300, 400, 500, 600, 700, 800, 900, 1000, 1100, 1200, 1300, 1400, 1500,
+ 1600, 1700, 1800, 1900, 2000, 2100
+ };
+ double[] v =
+ new double[] {
+ -1.2079272,
+ -0.01120245,
+ 1.1019456,
+ -0.52320362,
+ -0.35970289,
+ 0.1453591,
+ -0.45947892,
+ -1.0143219,
+ 0.81760821,
+ 0.5325646,
+ -0.29532424,
+ -0.1469335,
+ -0.12252526,
+ -0.67607713,
+ -0.16967308,
+ 0.8095585,
+ -0.78541944,
+ 0.03221141,
+ 0.31586886,
+ -0.41353356,
+ -0.21019539,
+ -0.0211206
+ };
+ config.setP1t(t[0]);
+ config.setP1v(v[0]);
+ config.setPnt(t[t.length - 1]);
+ config.setPnv(v[v.length - 1]);
+
+ for (int i = 0; i < t.length; i++) {
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, t[i], v[i]));
+ if ((i + 1) % 4 == 0) {
+ statement.execute("FLUSH");
+ }
+ }
+ statement.execute("FLUSH");
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+ }
+}