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 4d66b7b0d74317ee340b880aebeed2884e9f4f2b Author: Lei Rui <[email protected]> AuthorDate: Wed Sep 25 13:51:38 2024 +0800 test --- .../groupby/GroupByWithoutValueFilterDataSet.java | 234 +++---- .../groupby/LocalGroupByExecutorTri_LTTB.java | 97 +-- .../groupby/LocalGroupByExecutorTri_M4.java | 286 +++++--- .../groupby/LocalGroupByExecutorTri_MinMax.java | 237 ++++--- ...LocalGroupByExecutorTri_MinMaxPreselection.java | 759 +++++++++++---------- .../groupby/LocalGroupByExecutorTri_Uniform.java | 253 +++---- .../org/apache/iotdb/db/query/simpiece/FSW.java | 5 + .../db/query/simpiece/MySample_fsw_full2.java | 88 +-- .../simpiece/MySample_shrinkingcone_full2.java | 12 +- .../db/query/simpiece/MySample_simpiece_full2.java | 12 +- .../iotdb/db/query/simpiece/ShrinkingCone.java | 6 + .../apache/iotdb/db/query/simpiece/SimPiece.java | 4 + .../iotdb/db/query/simpiece/TimeSeriesReader.java | 17 +- .../iotdb/tsfile/read/common/ChunkSuit4Tri.java | 13 +- 14 files changed, 1030 insertions(+), 993 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 32eccb2076b..4e4e97f5307 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 @@ -209,120 +209,103 @@ public class GroupByWithoutValueFilterDataSet extends GroupByEngineDataSet { record = new RowRecord(0); StringBuilder series = new StringBuilder(); - try { - // First step: get the MinMax preselection result - List<Long> times = new ArrayList<>(); - List<Double> values = new ArrayList<>(); - GroupByExecutor executor = null; - for (Entry<PartialPath, GroupByExecutor> pathToExecutorEntry : pathExecutors.entrySet()) { - executor = pathToExecutorEntry.getValue(); // assume only one series here - break; - } - for (long localCurStartTime = startTime; - localCurStartTime + interval <= endTime; - // + interval to make the last bucket complete - // e.g, T=11,nout=3,interval=floor(11/3)=3, - // [0,3),[3,6),[6,9), no need incomplete [9,11) - // then the number of buckets must be Math.floor((endTime-startTime)/interval) - localCurStartTime += interval) { - // not change real curStartTime&curEndTime - // attention the returned aggregations need deep copy if using directly - List<AggregateResult> aggregations = - executor.calcResult( - localCurStartTime, localCurStartTime + interval, startTime, endTime, interval); - int c = 0; - for (AggregateResult aggregation : aggregations) { - // ATTENTION only take the first two aggregation fields, which are BPv[BPt], TPv[TPt] - // Each row correspond to (bucketLeftBound, minV[bottomT], maxV[topT]) of a MinMax bucket - MinMaxInfo minMaxInfo = (MinMaxInfo) aggregation.getResult(); - if (minMaxInfo == null) { - times.add(null); - values.add(null); - } else { - times.add(minMaxInfo.timestamp); - values.add((Double) minMaxInfo.val); - } - c++; - if (c >= 2) { - // ATTENTION only take the first two aggregation fields, which are BPv[BPt], TPv[TPt] - break; - } + // First step: get the MinMax preselection result + List<Long> times = new ArrayList<>(); + List<Double> values = new ArrayList<>(); + LocalGroupByExecutorTri_MinMax executor = null; + for (Entry<PartialPath, GroupByExecutor> pathToExecutorEntry : pathExecutors.entrySet()) { + executor = + (LocalGroupByExecutorTri_MinMax) + (pathToExecutorEntry.getValue()); // assume only one series here + break; + } + // get MinMax preselection times&values list + executor.calcResult( + startTime, startTime + interval, startTime, endTime, interval, times, values); + + // for (long localCurStartTime = startTime; + // localCurStartTime + interval <= endTime; + // // + interval to make the last bucket complete + // // e.g, T=11,nout=3,interval=floor(11/3)=3, + // // [0,3),[3,6),[6,9), no need incomplete [9,11) + // // then the number of buckets must be Math.floor((endTime-startTime)/interval) + // localCurStartTime += interval) { + // // not change real curStartTime&curEndTime + // // attention the returned aggregations need deep copy if using directly + // List<AggregateResult> aggregations = + // executor.calcResult( + // localCurStartTime, localCurStartTime + interval, startTime, endTime, + // interval); + // int c = 0; + // for (AggregateResult aggregation : aggregations) { + // // ATTENTION only take the first two aggregation fields, which are BPv[BPt], + // TPv[TPt] + // // Each row correspond to (bucketLeftBound, minV[bottomT], maxV[topT]) of a MinMax + // bucket + // MinMaxInfo minMaxInfo = (MinMaxInfo) aggregation.getResult(); + // if (minMaxInfo == null) { + // times.add(null); + // values.add(null); + // } else { + // times.add(minMaxInfo.timestamp); + // values.add((Double) minMaxInfo.val); + // } + // c++; + // if (c >= 2) { + // // ATTENTION only take the first two aggregation fields, which are BPv[BPt], + // TPv[TPt] + // break; + // } + // } + // } + + // Second step: apply LTTB on the MinMax preselection result + int N1 = (int) Math.floor((endTime * 1.0 - startTime) / interval); // MinMax桶数 + int N2 = N1 / (rps / divide); + series.append(p1v).append("[").append(p1t).append("]").append(","); + long lt = p1t; // left fixed t + double lv = p1v; // left fixed v + int currentBucket = 0; + for (; currentBucket < N2; currentBucket++) { + boolean emptyBucket = true; + for (int j = currentBucket * rps; j < (currentBucket + 1) * rps; j++) { + if (times.get(j) != null) { + emptyBucket = false; + break; } } - - // Second step: apply LTTB on the MinMax preselection result - int N1 = (int) Math.floor((endTime * 1.0 - startTime) / interval); // MinMax桶数 - int N2 = N1 / (rps / divide); - series.append(p1v).append("[").append(p1t).append("]").append(","); - long lt = p1t; // left fixed t - double lv = p1v; // left fixed v - int currentBucket = 0; - for (; currentBucket < N2; currentBucket++) { - boolean emptyBucket = true; - for (int j = currentBucket * rps; j < (currentBucket + 1) * rps; j++) { - if (times.get(j) != null) { - emptyBucket = false; - break; - } - } - if (!emptyBucket) { + if (!emptyBucket) { + break; + } + } + for (int nextBucket = currentBucket + 1; nextBucket < N2; nextBucket++) { + boolean emptyBucket = true; + for (int j = nextBucket * rps; j < (nextBucket + 1) * rps; j++) { + if (times.get(j) != null) { + emptyBucket = false; break; } } - for (int nextBucket = currentBucket + 1; nextBucket < N2; nextBucket++) { - boolean emptyBucket = true; - for (int j = nextBucket * rps; j < (nextBucket + 1) * rps; j++) { - if (times.get(j) != null) { - emptyBucket = false; - break; - } - } - if (emptyBucket) { - continue; - } + if (emptyBucket) { + continue; + } - double rt = 0; - double rv = 0; - int cnt = 0; - for (int j = nextBucket * rps; j < (nextBucket + 1) * rps; j++) { - if (times.get(j) != null) { - IOMonitor2.DCP_D_getAllSatisfiedPageData_traversedPointNum++; - rt += times.get(j); - rv += (double) values.get(j); - cnt++; - } - } - if (cnt == 0) { - throw new IOException("Empty bucket!"); - } - rt = rt / cnt; - rv = rv / cnt; - - double maxArea = -1; - long select_t = -1; - double select_v = -1; - for (int j = currentBucket * rps; j < (currentBucket + 1) * rps; j++) { - if (times.get(j) != null) { - IOMonitor2.DCP_D_getAllSatisfiedPageData_traversedPointNum++; - long t = times.get(j); - double v = values.get(j); - double area = IOMonitor2.calculateTri(lt, lv, t, v, rt, rv); - if (area > maxArea) { - maxArea = area; - select_t = t; - select_v = v; - } - } - } - if (select_t < 0) { - throw new IOException("something is wrong"); + double rt = 0; + double rv = 0; + int cnt = 0; + for (int j = nextBucket * rps; j < (nextBucket + 1) * rps; j++) { + if (times.get(j) != null) { + IOMonitor2.DCP_D_getAllSatisfiedPageData_traversedPointNum++; + rt += times.get(j); + rv += (double) values.get(j); + cnt++; } - series.append(select_v).append("[").append(select_t).append("]").append(","); - - currentBucket = nextBucket; - lt = select_t; - lv = select_v; } + if (cnt == 0) { + throw new IOException("Empty bucket!"); + } + rt = rt / cnt; + rv = rv / cnt; double maxArea = -1; long select_t = -1; @@ -332,7 +315,7 @@ public class GroupByWithoutValueFilterDataSet extends GroupByEngineDataSet { IOMonitor2.DCP_D_getAllSatisfiedPageData_traversedPointNum++; long t = times.get(j); double v = values.get(j); - double area = IOMonitor2.calculateTri(lt, lv, t, v, pnt, pnv); // 全局尾点作为右边固定点 + double area = IOMonitor2.calculateTri(lt, lv, t, v, rt, rv); if (area > maxArea) { maxArea = area; select_t = t; @@ -345,14 +328,35 @@ public class GroupByWithoutValueFilterDataSet extends GroupByEngineDataSet { } series.append(select_v).append("[").append(select_t).append("]").append(","); - series.append(pnv).append("[").append(pnt).append("]").append(","); - - record.addField(series, TSDataType.MIN_MAX_INT64); + currentBucket = nextBucket; + lt = select_t; + lv = select_v; + } - } catch (QueryProcessException e) { - logger.error("GroupByWithoutValueFilterDataSet execute has error", e); - throw new IOException(e.getMessage(), e); + double maxArea = -1; + long select_t = -1; + double select_v = -1; + for (int j = currentBucket * rps; j < (currentBucket + 1) * rps; j++) { + if (times.get(j) != null) { + IOMonitor2.DCP_D_getAllSatisfiedPageData_traversedPointNum++; + long t = times.get(j); + double v = values.get(j); + double area = IOMonitor2.calculateTri(lt, lv, t, v, pnt, pnv); // 全局尾点作为右边固定点 + if (area > maxArea) { + maxArea = area; + select_t = t; + select_v = v; + } + } + } + if (select_t < 0) { + throw new IOException("something is wrong"); } + series.append(select_v).append("[").append(select_t).append("]").append(","); + + series.append(pnv).append("[").append(pnt).append("]").append(","); + + record.addField(series, TSDataType.MIN_MAX_INT64); // in the end, make the next hasNextWithoutConstraint() false // as we already fetch all here @@ -432,8 +436,10 @@ public class GroupByWithoutValueFilterDataSet extends GroupByEngineDataSet { return new LocalGroupByExecutorTri_MinMax( path, allSensors, dataType, context, timeFilter, fileFilter, ascending); } else if (CONFIG.getEnableTri().equals("MinMaxLTTB")) { - return new LocalGroupByExecutorTri_MinMaxPreselection( + return new LocalGroupByExecutorTri_MinMax( path, allSensors, dataType, context, timeFilter, fileFilter, ascending); + // return new LocalGroupByExecutorTri_MinMaxPreselection( + // path, allSensors, dataType, context, timeFilter, fileFilter, ascending); } else if (CONFIG.getEnableTri().equals("M4")) { return new LocalGroupByExecutorTri_M4( path, allSensors, dataType, context, timeFilter, fileFilter, ascending); 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 cba1543aff8..319067017fa 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 @@ -31,8 +31,8 @@ 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.db.query.simpiece.TimeSeries; +import org.apache.iotdb.db.query.simpiece.TimeSeriesReader; 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; @@ -40,11 +40,9 @@ 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 java.io.IOException; -import java.nio.ByteBuffer; import java.util.ArrayList; import java.util.HashMap; import java.util.List; @@ -73,6 +71,8 @@ public class LocalGroupByExecutorTri_LTTB implements GroupByExecutor { private Filter timeFilter; + private TimeSeries timeSeries; + public LocalGroupByExecutorTri_LTTB( PartialPath path, Set<String> allSensors, @@ -115,7 +115,13 @@ public class LocalGroupByExecutorTri_LTTB implements GroupByExecutor { long endTime = groupByFilter.getEndTime(); long interval = groupByFilter.getInterval(); N1 = (int) Math.floor((endTime * 1.0 - startTime) / interval); // 分桶数 + + timeSeries = TimeSeriesReader.getTimeSeriesFromTsFiles(futureChunkList, startTime, endTime); + for (ChunkSuit4Tri chunkSuit4Tri : futureChunkList) { + // maintain which chunks fall into which buckets, facilitating LTTB calculating average and + // so on + // but ChunkSuit4Tri does not maintain data but only keep pointers to the timeseries list ChunkMetadata chunkMetadata = chunkSuit4Tri.chunkMetadata; long chunkMinTime = chunkMetadata.getStartTime(); long chunkMaxTime = chunkMetadata.getEndTime(); @@ -170,33 +176,32 @@ public class LocalGroupByExecutorTri_LTTB implements GroupByExecutor { long rightEndTime = startTime + (b + 2) * interval; int cnt = 0; 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. calculate avg - PageReader pageReader = chunkSuit4Tri.pageReader; - for (int j = 0; j < chunkSuit4Tri.chunkMetadata.getStatistics().getCount(); j++) { + // // 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. calculate avg + // PageReader pageReader = chunkSuit4Tri.pageReader; + for (int j = chunkSuit4Tri.globalStartInList; j < chunkSuit4Tri.globalEndInList; j++) { IOMonitor2.DCP_D_getAllSatisfiedPageData_traversedPointNum++; - long timestamp = pageReader.timeBuffer.getLong(j * 8); + // long timestamp = pageReader.timeBuffer.getLong(j * 8); + long timestamp = timeSeries.data.get(j).getTimestamp(); 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); + // ByteBuffer valueBuffer = pageReader.valueBuffer; + // double v = valueBuffer.getDouble(pageReader.timeBufferLength + j * 8); + double v = timeSeries.data.get(j).getValue(); rt += timestamp; rv += v; cnt++; @@ -217,33 +222,33 @@ public class LocalGroupByExecutorTri_LTTB implements GroupByExecutor { long localCurStartTime = startTime + (b) * interval; long localCurEndTime = startTime + (b + 1) * interval; 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; - int count = chunkSuit4Tri.chunkMetadata.getStatistics().getCount(); - int j; - for (j = 0; j < count; j++) { + // // 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; + // int count = chunkSuit4Tri.chunkMetadata.getStatistics().getCount(); + // int j; + // for (j = 0; j < count; j++) { + for (int j = chunkSuit4Tri.globalStartInList; j < chunkSuit4Tri.globalEndInList; j++) { IOMonitor2.DCP_D_getAllSatisfiedPageData_traversedPointNum++; - long timestamp = pageReader.timeBuffer.getLong(j * 8); + // long timestamp = pageReader.timeBuffer.getLong(j * 8); + long timestamp = timeSeries.data.get(j).getTimestamp(); 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); + // ByteBuffer valueBuffer = pageReader.valueBuffer; + // double v = valueBuffer.getDouble(pageReader.timeBufferLength + j * 8); + double v = timeSeries.data.get(j).getValue(); double area = IOMonitor2.calculateTri(lt, lv, timestamp, v, rt, rv); if (area > maxArea) { maxArea = area; diff --git a/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/LocalGroupByExecutorTri_M4.java b/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/LocalGroupByExecutorTri_M4.java index 9c20a9c1ff2..fac9cb286df 100644 --- a/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/LocalGroupByExecutorTri_M4.java +++ b/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/LocalGroupByExecutorTri_M4.java @@ -31,14 +31,11 @@ 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.query.simpiece.TimeSeries; +import org.apache.iotdb.db.query.simpiece.TimeSeriesReader; 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.DoubleStatistics; -import org.apache.iotdb.tsfile.file.metadata.statistics.FloatStatistics; -import org.apache.iotdb.tsfile.file.metadata.statistics.IntegerStatistics; -import org.apache.iotdb.tsfile.file.metadata.statistics.LongStatistics; 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; @@ -48,33 +45,31 @@ 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.Comparator; import java.util.List; -import java.util.ListIterator; import java.util.Set; public class LocalGroupByExecutorTri_M4 implements GroupByExecutor { private static final IoTDBConfig CONFIG = IoTDBDescriptor.getInstance().getConfig(); - private static final Logger M4_CHUNK_METADATA = LoggerFactory.getLogger("M4_CHUNK_METADATA"); + // 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<>(); + // private List<ChunkSuit4Tri> currentChunkList; + // private final List<ChunkSuit4Tri> futureChunkList = new ArrayList<>(); private Filter timeFilter; private final int N1; + private TimeSeries timeSeries; + public LocalGroupByExecutorTri_M4( PartialPath path, Set<String> allSensors, @@ -114,7 +109,7 @@ public class LocalGroupByExecutorTri_M4 implements GroupByExecutor { N1 = (int) Math.floor((endTime * 1.0 - startTime) / interval); // 分桶数 try { - // : this might be bad to load all chunk metadata at first + List<ChunkSuit4Tri> futureChunkList = new ArrayList<>(); futureChunkList.addAll(seriesReader.getAllChunkMetadatas4Tri()); // order futureChunkList by chunk startTime futureChunkList.sort( @@ -125,102 +120,104 @@ public class LocalGroupByExecutorTri_M4 implements GroupByExecutor { } }); - if (M4_CHUNK_METADATA.isDebugEnabled()) { - if (timeFilter instanceof GroupByFilter) { - M4_CHUNK_METADATA.debug( - "M4_QUERY_PARAM,{},{},{}", - ((GroupByFilter) timeFilter).getStartTime(), - ((GroupByFilter) timeFilter).getEndTime(), - ((GroupByFilter) timeFilter).getInterval()); - } - for (ChunkSuit4Tri ChunkSuit4Tri : futureChunkList) { - Statistics statistics = ChunkSuit4Tri.chunkMetadata.getStatistics(); - long FP_t = statistics.getStartTime(); - long LP_t = statistics.getEndTime(); - long BP_t = statistics.getBottomTimestamp(); - long TP_t = statistics.getTopTimestamp(); - switch (statistics.getType()) { - case INT32: - int FP_v_int = ((IntegerStatistics) statistics).getFirstValue(); - int LP_v_int = ((IntegerStatistics) statistics).getLastValue(); - int BP_v_int = ((IntegerStatistics) statistics).getMinValue(); - int TP_v_int = ((IntegerStatistics) statistics).getMaxValue(); - M4_CHUNK_METADATA.debug( - "M4_CHUNK_METADATA,{},{},{},{},{},{},{},{},{},{},{}", - FP_t, - LP_t, - BP_t, - TP_t, - FP_v_int, - LP_v_int, - BP_v_int, - TP_v_int, - ChunkSuit4Tri.chunkMetadata.getVersion(), - ChunkSuit4Tri.chunkMetadata.getOffsetOfChunkHeader(), - statistics.getCount()); - break; - case INT64: - long FP_v_long = ((LongStatistics) statistics).getFirstValue(); - long LP_v_long = ((LongStatistics) statistics).getLastValue(); - long BP_v_long = ((LongStatistics) statistics).getMinValue(); - long TP_v_long = ((LongStatistics) statistics).getMaxValue(); - M4_CHUNK_METADATA.debug( - "M4_CHUNK_METADATA,{},{},{},{},{},{},{},{},{},{},{}", - FP_t, - LP_t, - BP_t, - TP_t, - FP_v_long, - LP_v_long, - BP_v_long, - TP_v_long, - ChunkSuit4Tri.chunkMetadata.getVersion(), - ChunkSuit4Tri.chunkMetadata.getOffsetOfChunkHeader(), - statistics.getCount()); - break; - case FLOAT: - float FP_v_float = ((FloatStatistics) statistics).getFirstValue(); - float LP_v_float = ((FloatStatistics) statistics).getLastValue(); - float BP_v_float = ((FloatStatistics) statistics).getMinValue(); - float TP_v_float = ((FloatStatistics) statistics).getMaxValue(); - M4_CHUNK_METADATA.debug( - "M4_CHUNK_METADATA,{},{},{},{},{},{},{},{},{},{},{}", - FP_t, - LP_t, - BP_t, - TP_t, - FP_v_float, - LP_v_float, - BP_v_float, - TP_v_float, - ChunkSuit4Tri.chunkMetadata.getVersion(), - ChunkSuit4Tri.chunkMetadata.getOffsetOfChunkHeader(), - statistics.getCount()); - break; - case DOUBLE: - double FP_v_double = ((DoubleStatistics) statistics).getFirstValue(); - double LP_v_double = ((DoubleStatistics) statistics).getLastValue(); - double BP_v_double = ((DoubleStatistics) statistics).getMinValue(); - double TP_v_double = ((DoubleStatistics) statistics).getMaxValue(); - M4_CHUNK_METADATA.debug( - "M4_CHUNK_METADATA,{},{},{},{},{},{},{},{},{},{},{}", - FP_t, - LP_t, - BP_t, - TP_t, - FP_v_double, - LP_v_double, - BP_v_double, - TP_v_double, - ChunkSuit4Tri.chunkMetadata.getVersion(), - ChunkSuit4Tri.chunkMetadata.getOffsetOfChunkHeader(), - statistics.getCount()); - break; - default: - throw new QueryProcessException("unsupported data type!"); - } - } - } + timeSeries = TimeSeriesReader.getTimeSeriesFromTsFiles(futureChunkList, startTime, endTime); + + // if (M4_CHUNK_METADATA.isDebugEnabled()) { + // if (timeFilter instanceof GroupByFilter) { + // M4_CHUNK_METADATA.debug( + // "M4_QUERY_PARAM,{},{},{}", + // ((GroupByFilter) timeFilter).getStartTime(), + // ((GroupByFilter) timeFilter).getEndTime(), + // ((GroupByFilter) timeFilter).getInterval()); + // } + // for (ChunkSuit4Tri ChunkSuit4Tri : futureChunkList) { + // Statistics statistics = ChunkSuit4Tri.chunkMetadata.getStatistics(); + // long FP_t = statistics.getStartTime(); + // long LP_t = statistics.getEndTime(); + // long BP_t = statistics.getBottomTimestamp(); + // long TP_t = statistics.getTopTimestamp(); + // switch (statistics.getType()) { + // case INT32: + // int FP_v_int = ((IntegerStatistics) statistics).getFirstValue(); + // int LP_v_int = ((IntegerStatistics) statistics).getLastValue(); + // int BP_v_int = ((IntegerStatistics) statistics).getMinValue(); + // int TP_v_int = ((IntegerStatistics) statistics).getMaxValue(); + // M4_CHUNK_METADATA.debug( + // "M4_CHUNK_METADATA,{},{},{},{},{},{},{},{},{},{},{}", + // FP_t, + // LP_t, + // BP_t, + // TP_t, + // FP_v_int, + // LP_v_int, + // BP_v_int, + // TP_v_int, + // ChunkSuit4Tri.chunkMetadata.getVersion(), + // ChunkSuit4Tri.chunkMetadata.getOffsetOfChunkHeader(), + // statistics.getCount()); + // break; + // case INT64: + // long FP_v_long = ((LongStatistics) statistics).getFirstValue(); + // long LP_v_long = ((LongStatistics) statistics).getLastValue(); + // long BP_v_long = ((LongStatistics) statistics).getMinValue(); + // long TP_v_long = ((LongStatistics) statistics).getMaxValue(); + // M4_CHUNK_METADATA.debug( + // "M4_CHUNK_METADATA,{},{},{},{},{},{},{},{},{},{},{}", + // FP_t, + // LP_t, + // BP_t, + // TP_t, + // FP_v_long, + // LP_v_long, + // BP_v_long, + // TP_v_long, + // ChunkSuit4Tri.chunkMetadata.getVersion(), + // ChunkSuit4Tri.chunkMetadata.getOffsetOfChunkHeader(), + // statistics.getCount()); + // break; + // case FLOAT: + // float FP_v_float = ((FloatStatistics) statistics).getFirstValue(); + // float LP_v_float = ((FloatStatistics) statistics).getLastValue(); + // float BP_v_float = ((FloatStatistics) statistics).getMinValue(); + // float TP_v_float = ((FloatStatistics) statistics).getMaxValue(); + // M4_CHUNK_METADATA.debug( + // "M4_CHUNK_METADATA,{},{},{},{},{},{},{},{},{},{},{}", + // FP_t, + // LP_t, + // BP_t, + // TP_t, + // FP_v_float, + // LP_v_float, + // BP_v_float, + // TP_v_float, + // ChunkSuit4Tri.chunkMetadata.getVersion(), + // ChunkSuit4Tri.chunkMetadata.getOffsetOfChunkHeader(), + // statistics.getCount()); + // break; + // case DOUBLE: + // double FP_v_double = ((DoubleStatistics) statistics).getFirstValue(); + // double LP_v_double = ((DoubleStatistics) statistics).getLastValue(); + // double BP_v_double = ((DoubleStatistics) statistics).getMinValue(); + // double TP_v_double = ((DoubleStatistics) statistics).getMaxValue(); + // M4_CHUNK_METADATA.debug( + // "M4_CHUNK_METADATA,{},{},{},{},{},{},{},{},{},{},{}", + // FP_t, + // LP_t, + // BP_t, + // TP_t, + // FP_v_double, + // LP_v_double, + // BP_v_double, + // TP_v_double, + // ChunkSuit4Tri.chunkMetadata.getVersion(), + // ChunkSuit4Tri.chunkMetadata.getOffsetOfChunkHeader(), + // statistics.getCount()); + // break; + // default: + // throw new QueryProcessException("unsupported data type!"); + // } + // } + // } } catch (IOException e) { throw new QueryProcessException(e.getMessage()); @@ -235,6 +232,7 @@ public class LocalGroupByExecutorTri_M4 implements GroupByExecutor { results.add(aggrResult); } + /* private void getCurrentChunkListFromFutureChunkList(long curStartTime, long curEndTime) { // IOMonitor2.M4_LSM_status = Operation.M4_LSM_MERGE_M4_TIME_SPAN; @@ -267,6 +265,7 @@ public class LocalGroupByExecutorTri_M4 implements GroupByExecutor { } } } + */ @Override public List<AggregateResult> calcResult( @@ -281,11 +280,77 @@ public class LocalGroupByExecutorTri_M4 implements GroupByExecutor { series.append(CONFIG.getP1v()).append("[").append(CONFIG.getP1t()).append("]").append(","); - // Assume no empty buckets + int tmpLastReadPos = 0; + // use list without the notion of chunks for (int b = 0; b < N1; b++) { long localCurStartTime = startTime + (b) * interval; long localCurEndTime = startTime + (b + 1) * interval; + double minValue = Double.MAX_VALUE; + long bottomTime = -1; + double maxValue = -Double.MAX_VALUE; // Double.MIN_VALUE is positive so do not use it!!! + long topTime = -1; + long firstTime = -1; + double firstValue = 0; + long lastTime = -1; + double lastValue = 0; + + for (int i = tmpLastReadPos; i < timeSeries.data.size(); i++) { + IOMonitor2.DCP_D_getAllSatisfiedPageData_traversedPointNum++; + long timestamp = timeSeries.data.get(i).getTimestamp(); + if (timestamp < localCurStartTime) { + continue; + } else if (timestamp >= localCurEndTime) { + tmpLastReadPos = i; + break; + } else { + double v = timeSeries.data.get(i).getValue(); + if (firstTime < 0) { + firstTime = timestamp; + firstValue = v; + } + lastTime = timestamp; + lastValue = v; + if (v < minValue) { + minValue = v; + bottomTime = timestamp; + } + if (v > maxValue) { + maxValue = v; + topTime = timestamp; + } + } + } + + if (topTime >= 0) { + // minValue[bottomTime],maxValue[topTime],firstValue[firstTime],lastValue[lastTime] + series + .append(minValue) + .append("[") + .append(bottomTime) + .append("]") + .append(",") + .append(maxValue) + .append("[") + .append(topTime) + .append("]") + .append(",") + .append(firstValue) + .append("[") + .append(firstTime) + .append("]") + .append(",") + .append(lastValue) + .append("[") + .append(lastTime) + .append("]") + .append(","); + } else { + // minValue[bottomTime],maxValue[topTime],firstValue[firstTime],lastValue[lastTime] + series.append("null[null],null[null],null[null],null[null],"); + } + + /* getCurrentChunkListFromFutureChunkList(localCurStartTime, localCurEndTime); if (currentChunkList.size() == 0) { @@ -295,6 +360,7 @@ public class LocalGroupByExecutorTri_M4 implements GroupByExecutor { } calculateM4(currentChunkList, localCurStartTime, localCurEndTime, series); + */ } series.append(CONFIG.getPnv()).append("[").append(CONFIG.getPnt()).append("]").append(","); diff --git a/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/LocalGroupByExecutorTri_MinMax.java b/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/LocalGroupByExecutorTri_MinMax.java index 6d150e48976..9ac3e5c040c 100644 --- a/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/LocalGroupByExecutorTri_MinMax.java +++ b/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/LocalGroupByExecutorTri_MinMax.java @@ -31,14 +31,11 @@ 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.query.simpiece.TimeSeries; +import org.apache.iotdb.db.query.simpiece.TimeSeriesReader; 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.DoubleStatistics; -import org.apache.iotdb.tsfile.file.metadata.statistics.FloatStatistics; -import org.apache.iotdb.tsfile.file.metadata.statistics.IntegerStatistics; -import org.apache.iotdb.tsfile.file.metadata.statistics.LongStatistics; 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; @@ -48,33 +45,31 @@ 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.Comparator; import java.util.List; -import java.util.ListIterator; import java.util.Set; public class LocalGroupByExecutorTri_MinMax implements GroupByExecutor { private static final IoTDBConfig CONFIG = IoTDBDescriptor.getInstance().getConfig(); - private static final Logger M4_CHUNK_METADATA = LoggerFactory.getLogger("M4_CHUNK_METADATA"); + // 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<>(); + // private List<ChunkSuit4Tri> currentChunkList; + // private final List<ChunkSuit4Tri> futureChunkList = new ArrayList<>(); private Filter timeFilter; private final int N1; + private TimeSeries timeSeries; + public LocalGroupByExecutorTri_MinMax( PartialPath path, Set<String> allSensors, @@ -114,7 +109,7 @@ public class LocalGroupByExecutorTri_MinMax implements GroupByExecutor { N1 = (int) Math.floor((endTime * 1.0 - startTime) / interval); try { - // : this might be bad to load all chunk metadata at first + List<ChunkSuit4Tri> futureChunkList = new ArrayList<>(); futureChunkList.addAll(seriesReader.getAllChunkMetadatas4Tri()); // order futureChunkList by chunk startTime futureChunkList.sort( @@ -125,102 +120,7 @@ public class LocalGroupByExecutorTri_MinMax implements GroupByExecutor { } }); - if (M4_CHUNK_METADATA.isDebugEnabled()) { - if (timeFilter instanceof GroupByFilter) { - M4_CHUNK_METADATA.debug( - "M4_QUERY_PARAM,{},{},{}", - ((GroupByFilter) timeFilter).getStartTime(), - ((GroupByFilter) timeFilter).getEndTime(), - ((GroupByFilter) timeFilter).getInterval()); - } - for (ChunkSuit4Tri ChunkSuit4Tri : futureChunkList) { - Statistics statistics = ChunkSuit4Tri.chunkMetadata.getStatistics(); - long FP_t = statistics.getStartTime(); - long LP_t = statistics.getEndTime(); - long BP_t = statistics.getBottomTimestamp(); - long TP_t = statistics.getTopTimestamp(); - switch (statistics.getType()) { - case INT32: - int FP_v_int = ((IntegerStatistics) statistics).getFirstValue(); - int LP_v_int = ((IntegerStatistics) statistics).getLastValue(); - int BP_v_int = ((IntegerStatistics) statistics).getMinValue(); - int TP_v_int = ((IntegerStatistics) statistics).getMaxValue(); - M4_CHUNK_METADATA.debug( - "M4_CHUNK_METADATA,{},{},{},{},{},{},{},{},{},{},{}", - FP_t, - LP_t, - BP_t, - TP_t, - FP_v_int, - LP_v_int, - BP_v_int, - TP_v_int, - ChunkSuit4Tri.chunkMetadata.getVersion(), - ChunkSuit4Tri.chunkMetadata.getOffsetOfChunkHeader(), - statistics.getCount()); - break; - case INT64: - long FP_v_long = ((LongStatistics) statistics).getFirstValue(); - long LP_v_long = ((LongStatistics) statistics).getLastValue(); - long BP_v_long = ((LongStatistics) statistics).getMinValue(); - long TP_v_long = ((LongStatistics) statistics).getMaxValue(); - M4_CHUNK_METADATA.debug( - "M4_CHUNK_METADATA,{},{},{},{},{},{},{},{},{},{},{}", - FP_t, - LP_t, - BP_t, - TP_t, - FP_v_long, - LP_v_long, - BP_v_long, - TP_v_long, - ChunkSuit4Tri.chunkMetadata.getVersion(), - ChunkSuit4Tri.chunkMetadata.getOffsetOfChunkHeader(), - statistics.getCount()); - break; - case FLOAT: - float FP_v_float = ((FloatStatistics) statistics).getFirstValue(); - float LP_v_float = ((FloatStatistics) statistics).getLastValue(); - float BP_v_float = ((FloatStatistics) statistics).getMinValue(); - float TP_v_float = ((FloatStatistics) statistics).getMaxValue(); - M4_CHUNK_METADATA.debug( - "M4_CHUNK_METADATA,{},{},{},{},{},{},{},{},{},{},{}", - FP_t, - LP_t, - BP_t, - TP_t, - FP_v_float, - LP_v_float, - BP_v_float, - TP_v_float, - ChunkSuit4Tri.chunkMetadata.getVersion(), - ChunkSuit4Tri.chunkMetadata.getOffsetOfChunkHeader(), - statistics.getCount()); - break; - case DOUBLE: - double FP_v_double = ((DoubleStatistics) statistics).getFirstValue(); - double LP_v_double = ((DoubleStatistics) statistics).getLastValue(); - double BP_v_double = ((DoubleStatistics) statistics).getMinValue(); - double TP_v_double = ((DoubleStatistics) statistics).getMaxValue(); - M4_CHUNK_METADATA.debug( - "M4_CHUNK_METADATA,{},{},{},{},{},{},{},{},{},{},{}", - FP_t, - LP_t, - BP_t, - TP_t, - FP_v_double, - LP_v_double, - BP_v_double, - TP_v_double, - ChunkSuit4Tri.chunkMetadata.getVersion(), - ChunkSuit4Tri.chunkMetadata.getOffsetOfChunkHeader(), - statistics.getCount()); - break; - default: - throw new QueryProcessException("unsupported data type!"); - } - } - } + timeSeries = TimeSeriesReader.getTimeSeriesFromTsFiles(futureChunkList, startTime, endTime); } catch (IOException e) { throw new QueryProcessException(e.getMessage()); @@ -235,6 +135,7 @@ public class LocalGroupByExecutorTri_MinMax implements GroupByExecutor { results.add(aggrResult); } + /* private void getCurrentChunkListFromFutureChunkList(long curStartTime, long curEndTime) { // IOMonitor2.M4_LSM_status = Operation.M4_LSM_MERGE_M4_TIME_SPAN; @@ -267,6 +168,62 @@ public class LocalGroupByExecutorTri_MinMax implements GroupByExecutor { } } } + */ + + public void calcResult( + long curStartTime, + long curEndTime, + long startTime, + long endTime, + long interval, + List<Long> times, + List<Double> values) + throws IOException { + int tmpLastReadPos = 0; + // use list without the notion of chunks + for (int b = 0; b < N1; b++) { + long localCurStartTime = startTime + (b) * interval; // note local + long localCurEndTime = startTime + (b + 1) * interval; // note local + + double minValue = Double.MAX_VALUE; + long bottomTime = -1; + double maxValue = -Double.MAX_VALUE; // Double.MIN_VALUE is positive so do not use it!!! + long topTime = -1; + + for (int i = tmpLastReadPos; i < timeSeries.data.size(); i++) { + IOMonitor2.DCP_D_getAllSatisfiedPageData_traversedPointNum++; + long timestamp = timeSeries.data.get(i).getTimestamp(); + if (timestamp < localCurStartTime) { + continue; + } else if (timestamp >= localCurEndTime) { + tmpLastReadPos = i; + break; + } else { + double v = timeSeries.data.get(i).getValue(); + if (v < minValue) { + minValue = v; + bottomTime = timestamp; + } + if (v > maxValue) { + maxValue = v; + topTime = timestamp; + } + } + } + + if (topTime >= 0) { + times.add(bottomTime); + values.add(minValue); + times.add(topTime); + values.add(maxValue); + } else { + times.add(null); + values.add(null); + times.add(null); + values.add(null); + } + } + } @Override public List<AggregateResult> calcResult( @@ -281,11 +238,66 @@ public class LocalGroupByExecutorTri_MinMax implements GroupByExecutor { series.append(CONFIG.getP1v()).append("[").append(CONFIG.getP1t()).append("]").append(","); - // Assume no empty buckets + int tmpLastReadPos = 0; + // use list without the notion of chunks for (int b = 0; b < N1; b++) { - long localCurStartTime = startTime + (b) * interval; - long localCurEndTime = startTime + (b + 1) * interval; + long localCurStartTime = startTime + (b) * interval; // note local + long localCurEndTime = startTime + (b + 1) * interval; // note local + + double minValue = Double.MAX_VALUE; + long bottomTime = -1; + double maxValue = -Double.MAX_VALUE; // Double.MIN_VALUE is positive so do not use it!!! + long topTime = -1; + + for (int i = tmpLastReadPos; i < timeSeries.data.size(); i++) { + IOMonitor2.DCP_D_getAllSatisfiedPageData_traversedPointNum++; + long timestamp = timeSeries.data.get(i).getTimestamp(); + if (timestamp < localCurStartTime) { + continue; + } else if (timestamp >= localCurEndTime) { + tmpLastReadPos = i; + break; + } else { + double v = timeSeries.data.get(i).getValue(); + if (v < minValue) { + minValue = v; + bottomTime = timestamp; + } + if (v > maxValue) { + maxValue = v; + topTime = timestamp; + } + } + } + + if (topTime >= 0) { + series + .append(minValue) + .append("[") + .append(bottomTime) + .append("]") + .append(",") + .append(maxValue) + .append("[") + .append(topTime) + .append("]") + .append(","); + } else { + // empty bucket although statistics cover + series + .append("null") + .append("[") + .append("null") + .append("]") + .append(",") + .append("null") + .append("[") + .append("null") + .append("]") + .append(","); + } + /* getCurrentChunkListFromFutureChunkList(localCurStartTime, localCurEndTime); if (currentChunkList.size() == 0) { @@ -304,6 +316,7 @@ public class LocalGroupByExecutorTri_MinMax implements GroupByExecutor { } calculateMinMax(currentChunkList, localCurStartTime, localCurEndTime, series); + */ } series.append(CONFIG.getPnv()).append("[").append(CONFIG.getPnt()).append("]").append(","); diff --git a/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/LocalGroupByExecutorTri_MinMaxPreselection.java b/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/LocalGroupByExecutorTri_MinMaxPreselection.java index 655a0ca3d88..f0d3b307936 100644 --- a/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/LocalGroupByExecutorTri_MinMaxPreselection.java +++ b/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/LocalGroupByExecutorTri_MinMaxPreselection.java @@ -1,378 +1,381 @@ -/* - * 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.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.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; -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.DoubleStatistics; -import org.apache.iotdb.tsfile.file.metadata.statistics.FloatStatistics; -import org.apache.iotdb.tsfile.file.metadata.statistics.IntegerStatistics; -import org.apache.iotdb.tsfile.file.metadata.statistics.LongStatistics; -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; -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.Comparator; -import java.util.List; -import java.util.ListIterator; -import java.util.Set; - -public class LocalGroupByExecutorTri_MinMaxPreselection implements GroupByExecutor { - - 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<>(); - - private Filter timeFilter; - - public LocalGroupByExecutorTri_MinMaxPreselection( - 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); - - try { - // : this might be bad to load all chunk metadata at first - futureChunkList.addAll(seriesReader.getAllChunkMetadatas4Tri()); - // order futureChunkList by chunk startTime - futureChunkList.sort( - new Comparator<ChunkSuit4Tri>() { - public int compare(ChunkSuit4Tri o1, ChunkSuit4Tri o2) { - return ((Comparable) (o1.chunkMetadata.getStartTime())) - .compareTo(o2.chunkMetadata.getStartTime()); - } - }); - - if (M4_CHUNK_METADATA.isDebugEnabled()) { - if (timeFilter instanceof GroupByFilter) { - M4_CHUNK_METADATA.debug( - "M4_QUERY_PARAM,{},{},{}", - ((GroupByFilter) timeFilter).getStartTime(), - ((GroupByFilter) timeFilter).getEndTime(), - ((GroupByFilter) timeFilter).getInterval()); - } - for (ChunkSuit4Tri ChunkSuit4Tri : futureChunkList) { - Statistics statistics = ChunkSuit4Tri.chunkMetadata.getStatistics(); - long FP_t = statistics.getStartTime(); - long LP_t = statistics.getEndTime(); - long BP_t = statistics.getBottomTimestamp(); - long TP_t = statistics.getTopTimestamp(); - switch (statistics.getType()) { - case INT32: - int FP_v_int = ((IntegerStatistics) statistics).getFirstValue(); - int LP_v_int = ((IntegerStatistics) statistics).getLastValue(); - int BP_v_int = ((IntegerStatistics) statistics).getMinValue(); - int TP_v_int = ((IntegerStatistics) statistics).getMaxValue(); - M4_CHUNK_METADATA.debug( - "M4_CHUNK_METADATA,{},{},{},{},{},{},{},{},{},{},{}", - FP_t, - LP_t, - BP_t, - TP_t, - FP_v_int, - LP_v_int, - BP_v_int, - TP_v_int, - ChunkSuit4Tri.chunkMetadata.getVersion(), - ChunkSuit4Tri.chunkMetadata.getOffsetOfChunkHeader(), - statistics.getCount()); - break; - case INT64: - long FP_v_long = ((LongStatistics) statistics).getFirstValue(); - long LP_v_long = ((LongStatistics) statistics).getLastValue(); - long BP_v_long = ((LongStatistics) statistics).getMinValue(); - long TP_v_long = ((LongStatistics) statistics).getMaxValue(); - M4_CHUNK_METADATA.debug( - "M4_CHUNK_METADATA,{},{},{},{},{},{},{},{},{},{},{}", - FP_t, - LP_t, - BP_t, - TP_t, - FP_v_long, - LP_v_long, - BP_v_long, - TP_v_long, - ChunkSuit4Tri.chunkMetadata.getVersion(), - ChunkSuit4Tri.chunkMetadata.getOffsetOfChunkHeader(), - statistics.getCount()); - break; - case FLOAT: - float FP_v_float = ((FloatStatistics) statistics).getFirstValue(); - float LP_v_float = ((FloatStatistics) statistics).getLastValue(); - float BP_v_float = ((FloatStatistics) statistics).getMinValue(); - float TP_v_float = ((FloatStatistics) statistics).getMaxValue(); - M4_CHUNK_METADATA.debug( - "M4_CHUNK_METADATA,{},{},{},{},{},{},{},{},{},{},{}", - FP_t, - LP_t, - BP_t, - TP_t, - FP_v_float, - LP_v_float, - BP_v_float, - TP_v_float, - ChunkSuit4Tri.chunkMetadata.getVersion(), - ChunkSuit4Tri.chunkMetadata.getOffsetOfChunkHeader(), - statistics.getCount()); - break; - case DOUBLE: - double FP_v_double = ((DoubleStatistics) statistics).getFirstValue(); - double LP_v_double = ((DoubleStatistics) statistics).getLastValue(); - double BP_v_double = ((DoubleStatistics) statistics).getMinValue(); - double TP_v_double = ((DoubleStatistics) statistics).getMaxValue(); - M4_CHUNK_METADATA.debug( - "M4_CHUNK_METADATA,{},{},{},{},{},{},{},{},{},{},{}", - FP_t, - LP_t, - BP_t, - TP_t, - FP_v_double, - LP_v_double, - BP_v_double, - TP_v_double, - ChunkSuit4Tri.chunkMetadata.getVersion(), - ChunkSuit4Tri.chunkMetadata.getOffsetOfChunkHeader(), - statistics.getCount()); - break; - default: - throw new QueryProcessException("unsupported data type!"); - } - } - } - - } 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); - } - - 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) - throws IOException { - // clear result cache - for (AggregateResult result : results) { - result.reset(); - } - - getCurrentChunkListFromFutureChunkList(curStartTime, curEndTime); - - if (currentChunkList.size() == 0) { - return results; - } - - calculateMinMax(currentChunkList, curStartTime, curEndTime); - - 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 - - double minVal = Double.MAX_VALUE; - long bottomTime = -1; - double maxVal = -Double.MAX_VALUE; // Double.MIN_VALUE is positive so do not use it!!! - long topTime = -1; - - // 1. load page data if it hasn't been loaded - TSDataType dataType = chunkSuit4Tri.chunkMetadata.getDataType(); - if (dataType != TSDataType.DOUBLE) { - throw new UnSupportedDataTypeException(String.valueOf(dataType)); - } - 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; - int i; - for (i = chunkSuit4Tri.lastReadPos; i < count; i++) { - IOMonitor2.DCP_D_getAllSatisfiedPageData_traversedPointNum++; - 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; - double v = valueBuffer.getDouble(pageReader.timeBufferLength + i * 8); - if (v < minVal) { - minVal = v; - bottomTime = timestamp; - } - if (v > maxVal) { - maxVal = v; - topTime = timestamp; - } - } - } - // clear for heap space - if (i >= count) { - chunkSuit4Tri.pageReader = null; - } - // 4. update MinMax by traversing points fallen within this bucket - if (topTime >= 0) { - // 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) { - return false; - } - - @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"); - } -} +/// * +// * 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.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.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; +// 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.DoubleStatistics; +// import org.apache.iotdb.tsfile.file.metadata.statistics.FloatStatistics; +// import org.apache.iotdb.tsfile.file.metadata.statistics.IntegerStatistics; +// import org.apache.iotdb.tsfile.file.metadata.statistics.LongStatistics; +// 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; +// 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.Comparator; +// import java.util.List; +// import java.util.ListIterator; +// import java.util.Set; +// +// public class LocalGroupByExecutorTri_MinMaxPreselection implements GroupByExecutor { +// +// 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<>(); +// +// private Filter timeFilter; +// +// public LocalGroupByExecutorTri_MinMaxPreselection( +// 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); +// +// try { +// // : this might be bad to load all chunk metadata at first +// futureChunkList.addAll(seriesReader.getAllChunkMetadatas4Tri()); +// // order futureChunkList by chunk startTime +// futureChunkList.sort( +// new Comparator<ChunkSuit4Tri>() { +// public int compare(ChunkSuit4Tri o1, ChunkSuit4Tri o2) { +// return ((Comparable) (o1.chunkMetadata.getStartTime())) +// .compareTo(o2.chunkMetadata.getStartTime()); +// } +// }); +// +// if (M4_CHUNK_METADATA.isDebugEnabled()) { +// if (timeFilter instanceof GroupByFilter) { +// M4_CHUNK_METADATA.debug( +// "M4_QUERY_PARAM,{},{},{}", +// ((GroupByFilter) timeFilter).getStartTime(), +// ((GroupByFilter) timeFilter).getEndTime(), +// ((GroupByFilter) timeFilter).getInterval()); +// } +// for (ChunkSuit4Tri ChunkSuit4Tri : futureChunkList) { +// Statistics statistics = ChunkSuit4Tri.chunkMetadata.getStatistics(); +// long FP_t = statistics.getStartTime(); +// long LP_t = statistics.getEndTime(); +// long BP_t = statistics.getBottomTimestamp(); +// long TP_t = statistics.getTopTimestamp(); +// switch (statistics.getType()) { +// case INT32: +// int FP_v_int = ((IntegerStatistics) statistics).getFirstValue(); +// int LP_v_int = ((IntegerStatistics) statistics).getLastValue(); +// int BP_v_int = ((IntegerStatistics) statistics).getMinValue(); +// int TP_v_int = ((IntegerStatistics) statistics).getMaxValue(); +// M4_CHUNK_METADATA.debug( +// "M4_CHUNK_METADATA,{},{},{},{},{},{},{},{},{},{},{}", +// FP_t, +// LP_t, +// BP_t, +// TP_t, +// FP_v_int, +// LP_v_int, +// BP_v_int, +// TP_v_int, +// ChunkSuit4Tri.chunkMetadata.getVersion(), +// ChunkSuit4Tri.chunkMetadata.getOffsetOfChunkHeader(), +// statistics.getCount()); +// break; +// case INT64: +// long FP_v_long = ((LongStatistics) statistics).getFirstValue(); +// long LP_v_long = ((LongStatistics) statistics).getLastValue(); +// long BP_v_long = ((LongStatistics) statistics).getMinValue(); +// long TP_v_long = ((LongStatistics) statistics).getMaxValue(); +// M4_CHUNK_METADATA.debug( +// "M4_CHUNK_METADATA,{},{},{},{},{},{},{},{},{},{},{}", +// FP_t, +// LP_t, +// BP_t, +// TP_t, +// FP_v_long, +// LP_v_long, +// BP_v_long, +// TP_v_long, +// ChunkSuit4Tri.chunkMetadata.getVersion(), +// ChunkSuit4Tri.chunkMetadata.getOffsetOfChunkHeader(), +// statistics.getCount()); +// break; +// case FLOAT: +// float FP_v_float = ((FloatStatistics) statistics).getFirstValue(); +// float LP_v_float = ((FloatStatistics) statistics).getLastValue(); +// float BP_v_float = ((FloatStatistics) statistics).getMinValue(); +// float TP_v_float = ((FloatStatistics) statistics).getMaxValue(); +// M4_CHUNK_METADATA.debug( +// "M4_CHUNK_METADATA,{},{},{},{},{},{},{},{},{},{},{}", +// FP_t, +// LP_t, +// BP_t, +// TP_t, +// FP_v_float, +// LP_v_float, +// BP_v_float, +// TP_v_float, +// ChunkSuit4Tri.chunkMetadata.getVersion(), +// ChunkSuit4Tri.chunkMetadata.getOffsetOfChunkHeader(), +// statistics.getCount()); +// break; +// case DOUBLE: +// double FP_v_double = ((DoubleStatistics) statistics).getFirstValue(); +// double LP_v_double = ((DoubleStatistics) statistics).getLastValue(); +// double BP_v_double = ((DoubleStatistics) statistics).getMinValue(); +// double TP_v_double = ((DoubleStatistics) statistics).getMaxValue(); +// M4_CHUNK_METADATA.debug( +// "M4_CHUNK_METADATA,{},{},{},{},{},{},{},{},{},{},{}", +// FP_t, +// LP_t, +// BP_t, +// TP_t, +// FP_v_double, +// LP_v_double, +// BP_v_double, +// TP_v_double, +// ChunkSuit4Tri.chunkMetadata.getVersion(), +// ChunkSuit4Tri.chunkMetadata.getOffsetOfChunkHeader(), +// statistics.getCount()); +// break; +// default: +// throw new QueryProcessException("unsupported data type!"); +// } +// } +// } +// +// } 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); +// } +// +// 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) +// throws IOException { +// // clear result cache +// for (AggregateResult result : results) { +// result.reset(); +// } +// +// getCurrentChunkListFromFutureChunkList(curStartTime, curEndTime); +// +// if (currentChunkList.size() == 0) { +// return results; +// } +// +// calculateMinMax(currentChunkList, curStartTime, curEndTime); +// +// 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 +// +// double minVal = Double.MAX_VALUE; +// long bottomTime = -1; +// double maxVal = -Double.MAX_VALUE; // Double.MIN_VALUE is positive so do not use it!!! +// long topTime = -1; +// +// // 1. load page data if it hasn't been loaded +// TSDataType dataType = chunkSuit4Tri.chunkMetadata.getDataType(); +// if (dataType != TSDataType.DOUBLE) { +// throw new UnSupportedDataTypeException(String.valueOf(dataType)); +// } +// 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; +// int i; +// for (i = chunkSuit4Tri.lastReadPos; i < count; i++) { +// IOMonitor2.DCP_D_getAllSatisfiedPageData_traversedPointNum++; +// 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; +// double v = valueBuffer.getDouble(pageReader.timeBufferLength + i * 8); +// if (v < minVal) { +// minVal = v; +// bottomTime = timestamp; +// } +// if (v > maxVal) { +// maxVal = v; +// topTime = timestamp; +// } +// } +// } +// // clear for heap space +// if (i >= count) { +// chunkSuit4Tri.pageReader = null; +// } +// // 4. update MinMax by traversing points fallen within this bucket +// if (topTime >= 0) { +// // 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) +// { +// return false; +// } +// +// @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_Uniform.java b/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/LocalGroupByExecutorTri_Uniform.java index 1145376b544..e82e3410f09 100644 --- a/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/LocalGroupByExecutorTri_Uniform.java +++ b/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/LocalGroupByExecutorTri_Uniform.java @@ -31,50 +31,42 @@ 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.db.query.simpiece.TimeSeries; +import org.apache.iotdb.db.query.simpiece.TimeSeriesReader; 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.DoubleStatistics; -import org.apache.iotdb.tsfile.file.metadata.statistics.FloatStatistics; -import org.apache.iotdb.tsfile.file.metadata.statistics.IntegerStatistics; -import org.apache.iotdb.tsfile.file.metadata.statistics.LongStatistics; 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; 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.Comparator; import java.util.List; -import java.util.ListIterator; import java.util.Set; public class LocalGroupByExecutorTri_Uniform implements GroupByExecutor { private static final IoTDBConfig CONFIG = IoTDBDescriptor.getInstance().getConfig(); - private static final Logger M4_CHUNK_METADATA = LoggerFactory.getLogger("M4_CHUNK_METADATA"); + // 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<>(); + // private List<ChunkSuit4Tri> currentChunkList; + // private final List<ChunkSuit4Tri> futureChunkList = new ArrayList<>(); private Filter timeFilter; private final int N1; + private TimeSeries timeSeries; + public LocalGroupByExecutorTri_Uniform( PartialPath path, Set<String> allSensors, @@ -114,7 +106,7 @@ public class LocalGroupByExecutorTri_Uniform implements GroupByExecutor { N1 = (int) Math.floor((endTime * 1.0 - startTime) / interval); // 分桶数 try { - // : this might be bad to load all chunk metadata at first + List<ChunkSuit4Tri> futureChunkList = new ArrayList<>(); futureChunkList.addAll(seriesReader.getAllChunkMetadatas4Tri()); // order futureChunkList by chunk startTime futureChunkList.sort( @@ -125,102 +117,7 @@ public class LocalGroupByExecutorTri_Uniform implements GroupByExecutor { } }); - if (M4_CHUNK_METADATA.isDebugEnabled()) { - if (timeFilter instanceof GroupByFilter) { - M4_CHUNK_METADATA.debug( - "M4_QUERY_PARAM,{},{},{}", - ((GroupByFilter) timeFilter).getStartTime(), - ((GroupByFilter) timeFilter).getEndTime(), - ((GroupByFilter) timeFilter).getInterval()); - } - for (ChunkSuit4Tri ChunkSuit4Tri : futureChunkList) { - Statistics statistics = ChunkSuit4Tri.chunkMetadata.getStatistics(); - long FP_t = statistics.getStartTime(); - long LP_t = statistics.getEndTime(); - long BP_t = statistics.getBottomTimestamp(); - long TP_t = statistics.getTopTimestamp(); - switch (statistics.getType()) { - case INT32: - int FP_v_int = ((IntegerStatistics) statistics).getFirstValue(); - int LP_v_int = ((IntegerStatistics) statistics).getLastValue(); - int BP_v_int = ((IntegerStatistics) statistics).getMinValue(); - int TP_v_int = ((IntegerStatistics) statistics).getMaxValue(); - M4_CHUNK_METADATA.debug( - "M4_CHUNK_METADATA,{},{},{},{},{},{},{},{},{},{},{}", - FP_t, - LP_t, - BP_t, - TP_t, - FP_v_int, - LP_v_int, - BP_v_int, - TP_v_int, - ChunkSuit4Tri.chunkMetadata.getVersion(), - ChunkSuit4Tri.chunkMetadata.getOffsetOfChunkHeader(), - statistics.getCount()); - break; - case INT64: - long FP_v_long = ((LongStatistics) statistics).getFirstValue(); - long LP_v_long = ((LongStatistics) statistics).getLastValue(); - long BP_v_long = ((LongStatistics) statistics).getMinValue(); - long TP_v_long = ((LongStatistics) statistics).getMaxValue(); - M4_CHUNK_METADATA.debug( - "M4_CHUNK_METADATA,{},{},{},{},{},{},{},{},{},{},{}", - FP_t, - LP_t, - BP_t, - TP_t, - FP_v_long, - LP_v_long, - BP_v_long, - TP_v_long, - ChunkSuit4Tri.chunkMetadata.getVersion(), - ChunkSuit4Tri.chunkMetadata.getOffsetOfChunkHeader(), - statistics.getCount()); - break; - case FLOAT: - float FP_v_float = ((FloatStatistics) statistics).getFirstValue(); - float LP_v_float = ((FloatStatistics) statistics).getLastValue(); - float BP_v_float = ((FloatStatistics) statistics).getMinValue(); - float TP_v_float = ((FloatStatistics) statistics).getMaxValue(); - M4_CHUNK_METADATA.debug( - "M4_CHUNK_METADATA,{},{},{},{},{},{},{},{},{},{},{}", - FP_t, - LP_t, - BP_t, - TP_t, - FP_v_float, - LP_v_float, - BP_v_float, - TP_v_float, - ChunkSuit4Tri.chunkMetadata.getVersion(), - ChunkSuit4Tri.chunkMetadata.getOffsetOfChunkHeader(), - statistics.getCount()); - break; - case DOUBLE: - double FP_v_double = ((DoubleStatistics) statistics).getFirstValue(); - double LP_v_double = ((DoubleStatistics) statistics).getLastValue(); - double BP_v_double = ((DoubleStatistics) statistics).getMinValue(); - double TP_v_double = ((DoubleStatistics) statistics).getMaxValue(); - M4_CHUNK_METADATA.debug( - "M4_CHUNK_METADATA,{},{},{},{},{},{},{},{},{},{},{}", - FP_t, - LP_t, - BP_t, - TP_t, - FP_v_double, - LP_v_double, - BP_v_double, - TP_v_double, - ChunkSuit4Tri.chunkMetadata.getVersion(), - ChunkSuit4Tri.chunkMetadata.getOffsetOfChunkHeader(), - statistics.getCount()); - break; - default: - throw new QueryProcessException("unsupported data type!"); - } - } - } + timeSeries = TimeSeriesReader.getTimeSeriesFromTsFiles(futureChunkList, startTime, endTime); } catch (IOException e) { throw new QueryProcessException(e.getMessage()); @@ -235,39 +132,24 @@ public class LocalGroupByExecutorTri_Uniform 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 - } - } - } - + /** + * private void getCurrentChunkListFromFutureChunkList(long curStartTime, long curEndTime) { // + * IOMonitor2.M4_LSM_status = Operation.M4_LSM_MERGE_M4_TIME_SPAN; + * + * <p>// empty currentChunkList currentChunkList = new ArrayList<>(); + * + * <p>// 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) @@ -281,11 +163,42 @@ public class LocalGroupByExecutorTri_Uniform implements GroupByExecutor { series.append(CONFIG.getP1v()).append("[").append(CONFIG.getP1t()).append("]").append(","); - // Assume no empty buckets + int tmpLastReadPos = 0; + // use list without the notion of chunks for (int b = 0; b < N1; b++) { - long localCurStartTime = startTime + (b) * interval; - long localCurEndTime = startTime + (b + 1) * interval; + long localCurStartTime = startTime + (b) * interval; // note local + long localCurEndTime = startTime + (b + 1) * interval; // note local + + long firstTime = -1; + double firstValue = 0; + + for (int i = tmpLastReadPos; i < timeSeries.data.size(); i++) { + IOMonitor2.DCP_D_getAllSatisfiedPageData_traversedPointNum++; + long timestamp = timeSeries.data.get(i).getTimestamp(); + if (timestamp < localCurStartTime) { + continue; + } else if (timestamp >= localCurEndTime) { + tmpLastReadPos = i; + break; + } else { + if (firstTime < 0) { + double v = timeSeries.data.get(i).getValue(); + firstTime = timestamp; + firstValue = v; + } + } + } + if (firstTime >= 0) { + // minValue[bottomTime],maxValue[topTime],firstValue[firstTime],lastValue[lastTime] + series.append(firstValue).append("[").append(firstTime).append("]").append(","); + } else { + // empty bucket although statistics cover + // minValue[bottomTime],maxValue[topTime],firstValue[firstTime],lastValue[lastTime] + series.append("null[null],"); + } + + /* getCurrentChunkListFromFutureChunkList(localCurStartTime, localCurEndTime); if (currentChunkList.size() == 0) { @@ -295,6 +208,7 @@ public class LocalGroupByExecutorTri_Uniform implements GroupByExecutor { } calculateFirst(currentChunkList, localCurStartTime, localCurEndTime, series); + */ } series.append(CONFIG.getPnv()).append("[").append(CONFIG.getPnt()).append("]").append(","); @@ -302,6 +216,9 @@ public class LocalGroupByExecutorTri_Uniform implements GroupByExecutor { MinValueAggrResult minValueAggrResult = (MinValueAggrResult) results.get(0); minValueAggrResult.updateResult(new MinMaxInfo<>(series.toString(), 0)); + System.out.println( + "traversed points=" + IOMonitor2.DCP_D_getAllSatisfiedPageData_traversedPointNum); + return results; } @@ -330,22 +247,24 @@ public class LocalGroupByExecutorTri_Uniform implements GroupByExecutor { if (dataType != TSDataType.DOUBLE) { throw new UnSupportedDataTypeException(String.valueOf(dataType)); } - 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; + // 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; int i; - for (i = chunkSuit4Tri.lastReadPos; i < count; i++) { + for (i = chunkSuit4Tri.lastReadPos; i < chunkSuit4Tri.globalEndInList; i++) { IOMonitor2.DCP_D_getAllSatisfiedPageData_traversedPointNum++; - long timestamp = pageReader.timeBuffer.getLong(i * 8); + long timestamp = timeSeries.data.get(i).getTimestamp(); + // 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) @@ -353,21 +272,23 @@ public class LocalGroupByExecutorTri_Uniform implements GroupByExecutor { } else if (timestamp >= curEndTime) { // 3. traverse until the first point fallen right this bucket, also remember to update // lastReadPos - chunkSuit4Tri.lastReadPos = i; + chunkSuit4Tri.lastReadPos = i; // note global pos in list break; } else { // 4. update MinMax by traversing points fallen within this bucket - ByteBuffer valueBuffer = pageReader.valueBuffer; - double v = valueBuffer.getDouble(pageReader.timeBufferLength + i * 8); + // ByteBuffer valueBuffer = pageReader.valueBuffer; + // double v = valueBuffer.getDouble(pageReader.timeBufferLength + i * 8); + double v = timeSeries.data.get(i).getValue(); if (firstTime < 0) { firstTime = timestamp; firstValue = v; - break; + // break; } } } } } + if (firstTime >= 0) { // minValue[bottomTime],maxValue[topTime],firstValue[firstTime],lastValue[lastTime] series.append(firstValue).append("[").append(firstTime).append("]").append(","); diff --git a/server/src/main/java/org/apache/iotdb/db/query/simpiece/FSW.java b/server/src/main/java/org/apache/iotdb/db/query/simpiece/FSW.java index 057e7c29607..7d9d9c3038c 100644 --- a/server/src/main/java/org/apache/iotdb/db/query/simpiece/FSW.java +++ b/server/src/main/java/org/apache/iotdb/db/query/simpiece/FSW.java @@ -19,6 +19,8 @@ package org.apache.iotdb.db.query.simpiece; +import org.apache.iotdb.tsfile.read.common.IOMonitor2; + import java.util.ArrayList; import java.util.List; @@ -50,6 +52,7 @@ public class FSW { double lowSlope = Double.NEGATIVE_INFINITY; while (i < length - 1) { + IOMonitor2.DCP_D_getAllSatisfiedPageData_traversedPointNum++; i++; upSlope = Math.min( @@ -82,6 +85,7 @@ public class FSW { // deal with the last segment if (segmentPoint.get(seg_no)[1] < length - 1) { + IOMonitor2.DCP_D_getAllSatisfiedPageData_traversedPointNum++; seg_no += 1; segmentPoint.add(new int[] {seg_no, length - 1}); } @@ -90,6 +94,7 @@ public class FSW { for (int[] ints : segmentPoint) { result.add(points.get(ints[1])); } + return result; } } diff --git a/server/src/main/java/org/apache/iotdb/db/query/simpiece/MySample_fsw_full2.java b/server/src/main/java/org/apache/iotdb/db/query/simpiece/MySample_fsw_full2.java index c6db83380b1..e8d8cb58a44 100644 --- a/server/src/main/java/org/apache/iotdb/db/query/simpiece/MySample_fsw_full2.java +++ b/server/src/main/java/org/apache/iotdb/db/query/simpiece/MySample_fsw_full2.java @@ -35,48 +35,48 @@ public class MySample_fsw_full2 { public static void main(String[] args) { String fileDir = "D:\\desktop\\NISTPV\\"; // do not change the order of datasets below, as the output is used in exp bash - String[] datasetNameList = new String[]{"WindSpeed", "Qloss", "Pyra1", "RTD"}; + String[] datasetNameList = new String[] {"WindSpeed", "Qloss", "Pyra1", "RTD"}; int[] noutList = - new int[]{ - 320, 400, 480, 580, 720, 960, 1200, 1600, 2000, 2400, 3000, 3600, 4000, 4400, 5000 + new int[] { + 320, 400, 480, 580, 720, 960, 1200, 1600, 2000, 2400, 3000, 3600, 4000, 4400, 5000 }; double[][] epsilonArray = { - { - 9.618271827697754, - 8.991525650024414, - 8.375, - 7.99461555480957, - 7.615596771240234, - 7.172431945800781, - 6.877520561218262, - 6.408135414123535, - 6.219999313354492, - 5.977571487426758, - 5.679183006286621, - 5.499468803405762, - 5.387459754943848, - 5.299999237060547, - 5.1635847091674805, - }, - { - 9.9945068359375E-4, 9.984970092773438E-4, 9.965896606445312E-4, 9.937286376953125E-4, - 9.899139404296875E-4, 9.813308715820312E-4, 9.6893310546875E-4, 9.42230224609375E-4, - 8.983612060546875E-4, 8.268356323242188E-4, 6.685256958007812E-4, 5.197525024414062E-4, - 4.99725341796875E-4, 4.987716674804688E-4, 4.8732757568359375E-4, - }, - { - 440.0235958099365, 423.5567502975464, 405.7711305618286, 396.31347465515137, - 380.6776990890503, 358.0119905471802, 336.5447692871094, 303.85207748413086, - 279.7541666030884, 257.3554916381836, 232.57367420196533, 211.35449981689453, - 200.9514446258545, 192.0128345489502, 178.28646087646484, - }, - { - 9.295397758483887, 7.5366668701171875, 6.473535537719727, 5.6244354248046875, - 4.713288307189941, 3.7258691787719727, 3.078885078430176, 2.4637460708618164, - 2.0368423461914062, 1.7683038711547852, 1.4741735458374023, 1.266657829284668, - 1.1668891906738281, 1.071258544921875, 0.9608230590820312, - } + { + 9.618271827697754, + 8.991525650024414, + 8.375, + 7.99461555480957, + 7.615596771240234, + 7.172431945800781, + 6.877520561218262, + 6.408135414123535, + 6.219999313354492, + 5.977571487426758, + 5.679183006286621, + 5.499468803405762, + 5.387459754943848, + 5.299999237060547, + 5.1635847091674805, + }, + { + 9.9945068359375E-4, 9.984970092773438E-4, 9.965896606445312E-4, 9.937286376953125E-4, + 9.899139404296875E-4, 9.813308715820312E-4, 9.6893310546875E-4, 9.42230224609375E-4, + 8.983612060546875E-4, 8.268356323242188E-4, 6.685256958007812E-4, 5.197525024414062E-4, + 4.99725341796875E-4, 4.987716674804688E-4, 4.8732757568359375E-4, + }, + { + 440.0235958099365, 423.5567502975464, 405.7711305618286, 396.31347465515137, + 380.6776990890503, 358.0119905471802, 336.5447692871094, 303.85207748413086, + 279.7541666030884, 257.3554916381836, 232.57367420196533, 211.35449981689453, + 200.9514446258545, 192.0128345489502, 178.28646087646484, + }, + { + 9.295397758483887, 7.5366668701171875, 6.473535537719727, 5.6244354248046875, + 4.713288307189941, 3.7258691787719727, 3.078885078430176, 2.4637460708618164, + 2.0368423461914062, 1.7683038711547852, 1.4741735458374023, 1.266657829284668, + 1.1668891906738281, 1.071258544921875, 0.9608230590820312, + } }; // double[][] epsilonArray = new double[datasetNameList.length][]; @@ -135,12 +135,12 @@ public class MySample_fsw_full2 { } } -// for (int i = 0; i < epsilonArray.length; i++) { -// for (int j = 0; j < epsilonArray[i].length; j++) { -// System.out.print(epsilonArray[i][j] + ","); -// } -// System.out.println(); -// } + // for (int i = 0; i < epsilonArray.length; i++) { + // for (int j = 0; j < epsilonArray[i].length; j++) { + // System.out.print(epsilonArray[i][j] + ","); + // } + // System.out.println(); + // } // do not change name of the output file, as the output is used in exp bash try (FileWriter writer = new FileWriter("epsilonArray_fsw.txt")) { diff --git a/server/src/main/java/org/apache/iotdb/db/query/simpiece/MySample_shrinkingcone_full2.java b/server/src/main/java/org/apache/iotdb/db/query/simpiece/MySample_shrinkingcone_full2.java index b085f74bf8c..25e44198e64 100644 --- a/server/src/main/java/org/apache/iotdb/db/query/simpiece/MySample_shrinkingcone_full2.java +++ b/server/src/main/java/org/apache/iotdb/db/query/simpiece/MySample_shrinkingcone_full2.java @@ -124,12 +124,12 @@ public class MySample_shrinkingcone_full2 { } } -// for (int i = 0; i < epsilonArray.length; i++) { // 遍历行 -// for (int j = 0; j < epsilonArray[i].length; j++) { // 遍历列 -// System.out.print(epsilonArray[i][j] + ","); -// } -// System.out.println(); -// } + // for (int i = 0; i < epsilonArray.length; i++) { // 遍历行 + // for (int j = 0; j < epsilonArray[i].length; j++) { // 遍历列 + // System.out.print(epsilonArray[i][j] + ","); + // } + // System.out.println(); + // } // do not change name of the output file, as the output is used in exp bash try (FileWriter writer = new FileWriter("epsilonArray_sc.txt")) { diff --git a/server/src/main/java/org/apache/iotdb/db/query/simpiece/MySample_simpiece_full2.java b/server/src/main/java/org/apache/iotdb/db/query/simpiece/MySample_simpiece_full2.java index d0cb0bd7278..e0d3eaccdf2 100644 --- a/server/src/main/java/org/apache/iotdb/db/query/simpiece/MySample_simpiece_full2.java +++ b/server/src/main/java/org/apache/iotdb/db/query/simpiece/MySample_simpiece_full2.java @@ -145,12 +145,12 @@ public class MySample_simpiece_full2 { } } -// for (int i = 0; i < epsilonArray.length; i++) { // 遍历行 -// for (int j = 0; j < epsilonArray[i].length; j++) { // 遍历列 -// System.out.print(epsilonArray[i][j] + ","); -// } -// System.out.println(); -// } + // for (int i = 0; i < epsilonArray.length; i++) { // 遍历行 + // for (int j = 0; j < epsilonArray[i].length; j++) { // 遍历列 + // System.out.print(epsilonArray[i][j] + ","); + // } + // System.out.println(); + // } // do not change name of the output file, as the output is used in exp bash try (FileWriter writer = new FileWriter("epsilonArray_simpiece.txt")) { diff --git a/server/src/main/java/org/apache/iotdb/db/query/simpiece/ShrinkingCone.java b/server/src/main/java/org/apache/iotdb/db/query/simpiece/ShrinkingCone.java index a5977068a69..33be3a2323d 100644 --- a/server/src/main/java/org/apache/iotdb/db/query/simpiece/ShrinkingCone.java +++ b/server/src/main/java/org/apache/iotdb/db/query/simpiece/ShrinkingCone.java @@ -19,6 +19,8 @@ package org.apache.iotdb.db.query.simpiece; +import org.apache.iotdb.tsfile.read.common.IOMonitor2; + import java.util.ArrayList; import java.util.List; @@ -42,12 +44,14 @@ public class ShrinkingCone { // init the first segment int sp = 0; int i = 1; + IOMonitor2.DCP_D_getAllSatisfiedPageData_traversedPointNum++; double vsp = points.get(sp).getValue(); double dx = points.get(i).getTimestamp() - points.get(sp).getTimestamp(); double upSlope = (p_upper[i] - vsp) / dx; double lowSlope = (p_lower[i] - vsp) / dx; while (i < length - 1) { + IOMonitor2.DCP_D_getAllSatisfiedPageData_traversedPointNum++; i++; vsp = points.get(sp).getValue(); // the value of the start point dx = points.get(i).getTimestamp() - points.get(sp).getTimestamp(); // time distance @@ -64,6 +68,7 @@ public class ShrinkingCone { result.add(points.get(i - 1)); // begin new segment + IOMonitor2.DCP_D_getAllSatisfiedPageData_traversedPointNum++; sp = i - 1; // joint style vsp = points.get(sp).getValue(); // note sp has changed dx = points.get(i).getTimestamp() - points.get(sp).getTimestamp(); // note sp has changed @@ -73,6 +78,7 @@ public class ShrinkingCone { } // write last point + IOMonitor2.DCP_D_getAllSatisfiedPageData_traversedPointNum++; result.add(points.get(points.size() - 1)); return result; diff --git a/server/src/main/java/org/apache/iotdb/db/query/simpiece/SimPiece.java b/server/src/main/java/org/apache/iotdb/db/query/simpiece/SimPiece.java index a7c491f81c9..39e185c0062 100644 --- a/server/src/main/java/org/apache/iotdb/db/query/simpiece/SimPiece.java +++ b/server/src/main/java/org/apache/iotdb/db/query/simpiece/SimPiece.java @@ -24,6 +24,7 @@ package org.apache.iotdb.db.query.simpiece; import org.apache.iotdb.db.query.simpiece.Encoding.FloatEncoder; import org.apache.iotdb.db.query.simpiece.Encoding.UIntEncoder; import org.apache.iotdb.db.query.simpiece.Encoding.VariableByteEncoder; +import org.apache.iotdb.tsfile.read.common.IOMonitor2; import java.io.ByteArrayInputStream; import java.io.ByteArrayOutputStream; @@ -71,6 +72,8 @@ public class SimPiece { } for (int idx = startIdx + 2; idx < points.size(); idx++) { + IOMonitor2.DCP_D_getAllSatisfiedPageData_traversedPointNum++; + double upValue = points.get(idx).getValue() + epsilon; double downValue = points.get(idx).getValue() - epsilon; @@ -110,6 +113,7 @@ public class SimPiece { Comparator.comparingDouble(SimPieceSegment::getB) .thenComparingDouble(SimPieceSegment::getA)); for (int i = 0; i < segments.size(); i++) { + IOMonitor2.DCP_D_getAllSatisfiedPageData_traversedPointNum++; if (b != segments.get(i).getB()) { if (timestamps.size() == 1) mergedSegments.add(new SimPieceSegment(timestamps.get(0), aMinTemp, aMaxTemp, b)); diff --git a/server/src/main/java/org/apache/iotdb/db/query/simpiece/TimeSeriesReader.java b/server/src/main/java/org/apache/iotdb/db/query/simpiece/TimeSeriesReader.java index 65b828cb1b1..29eb0003d6d 100644 --- a/server/src/main/java/org/apache/iotdb/db/query/simpiece/TimeSeriesReader.java +++ b/server/src/main/java/org/apache/iotdb/db/query/simpiece/TimeSeriesReader.java @@ -22,9 +22,10 @@ package org.apache.iotdb.db.query.simpiece; 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.read.common.ChunkSuit4Tri; -import org.apache.iotdb.tsfile.read.common.IOMonitor2; import org.apache.iotdb.tsfile.read.reader.page.PageReader; import java.io.BufferedReader; @@ -47,7 +48,13 @@ public class TimeSeriesReader { double max = Double.MIN_VALUE; double min = Double.MAX_VALUE; + int start = 0; + for (ChunkSuit4Tri chunkSuit4Tri : chunkSuit4TriList) { + TSDataType dataType = chunkSuit4Tri.chunkMetadata.getDataType(); + if (dataType != TSDataType.DOUBLE) { + throw new UnSupportedDataTypeException(String.valueOf(dataType)); + } ChunkMetadata chunkMetadata = chunkSuit4Tri.chunkMetadata; long chunkMinTime = chunkMetadata.getStartTime(); long chunkMaxTime = chunkMetadata.getEndTime(); @@ -56,12 +63,13 @@ public class TimeSeriesReader { } else if (chunkMinTime >= endTime) { break; } else { - PageReader pageReader = + chunkSuit4Tri.globalStartInList = start; // pointer start + chunkSuit4Tri.lastReadPos = start; // note this means global pos in the list + PageReader pageReader = // note this pageReader and its buffer is not maintained in memory FileLoaderUtils.loadPageReaderList4CPV( chunkSuit4Tri.chunkMetadata, null); // note do not assign to chunkSuit4Tri.pageReader for (int j = 0; j < chunkSuit4Tri.chunkMetadata.getStatistics().getCount(); j++) { - IOMonitor2.DCP_D_getAllSatisfiedPageData_traversedPointNum++; long timestamp = pageReader.timeBuffer.getLong(j * 8); if (timestamp < startTime) { continue; @@ -73,8 +81,10 @@ public class TimeSeriesReader { ts.add(new Point(timestamp, value)); max = Math.max(max, value); min = Math.min(min, value); + start++; } } + chunkSuit4Tri.globalEndInList = start; // pointer end } } return new TimeSeries(ts, max - min); @@ -101,7 +111,6 @@ public class TimeSeriesReader { chunkSuit4Tri.chunkMetadata, null); // note do not assign to chunkSuit4Tri.pageReader for (int j = 0; j < chunkSuit4Tri.chunkMetadata.getStatistics().getCount(); j++) { - IOMonitor2.DCP_D_getAllSatisfiedPageData_traversedPointNum++; long timestamp = pageReader.timeBuffer.getLong(j * 8); if (timestamp < startTime) { continue; diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/ChunkSuit4Tri.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/ChunkSuit4Tri.java index 5e2388debde..95b5ea21cf9 100644 --- a/tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/ChunkSuit4Tri.java +++ b/tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/ChunkSuit4Tri.java @@ -24,17 +24,16 @@ import org.apache.iotdb.tsfile.read.reader.page.PageReader; public class ChunkSuit4Tri { - public ChunkMetadata chunkMetadata; // fixed info, including version, dataType, stepRegress + public int globalStartInList = 0; // close, pointer to pos in timeseries list + public int globalEndInList = 0; // open, pointer to pos in timeseries list // dynamic maintained globally, starting from 0, incremental, never decrease. - // only used in LocalGroupByExecutorTri_MinMax as it never reads backward - public int lastReadPos = 0; + public int lastReadPos = 0; // pointer to pos in timeseries list + + public ChunkMetadata chunkMetadata; // fixed info, including version, dataType, stepRegress - // TODO ATTENTION: YOU HAVE TO ENSURE THAT THERE IS ONLY ONE PAGE IN A CHUNK, + // 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! public PageReader pageReader; // bears fixed plain timeBuffer and valueBuffer // pageReader does not refer to the same deleteInterval as those in chunkMetadata // after chunkMetadata executes insertIntoSortedDeletions
