This is an automated email from the ASF dual-hosted git repository. qiaojialin pushed a commit to branch series_reader_doc in repository https://gitbox.apache.org/repos/asf/incubator-iotdb.git
commit 30421a9b8a9b31966dafa93e4b02e66894ec0dd9 Author: qiaojialin <[email protected]> AuthorDate: Tue Mar 3 11:29:08 2020 +0800 update data query doc --- .../SystemDesign/5-DataQuery/2-SeriesReader.md | 410 ++++++++++----------- .../SystemDesign/5-DataQuery/4-AggregationQuery.md | 12 +- .../SystemDesign/5-DataQuery/5-GroupByQuery.md | 17 +- .../db/query/reader/series/IAggregateReader.java | 2 +- .../query/reader/series/SeriesAggregateReader.java | 4 +- .../reader/series/SeriesRawDataPointReader.java | 3 + .../iotdb/db/query/reader/series/SeriesReader.java | 91 ++--- .../org/apache/iotdb/db/utils/FileLoaderUtils.java | 40 +- 8 files changed, 269 insertions(+), 310 deletions(-) diff --git a/docs/Documentation-CHN/SystemDesign/5-DataQuery/2-SeriesReader.md b/docs/Documentation-CHN/SystemDesign/5-DataQuery/2-SeriesReader.md index eb6c5da..bcc59f3 100644 --- a/docs/Documentation-CHN/SystemDesign/5-DataQuery/2-SeriesReader.md +++ b/docs/Documentation-CHN/SystemDesign/5-DataQuery/2-SeriesReader.md @@ -23,59 +23,29 @@ ## 设计原理 -IoTDB server 模块共提供 4 种不同形式的针对单个时间序列的读取接口，以支持不同形式的查询。 +IoTDB server 模块共提供 3 种不同形式的针对单个时间序列的读取接口，以支持不同形式的查询。 -* 按批遍历形式，返回 BatchData（主要用于不带值过滤的原始数据查询） +* 原始数据查询接口，返回 BatchData，可带时间过滤条件或值过滤条件，两种过滤不可同时存在。 * 聚合查询接口 （主要用于聚合查询和降采样查询） -* 单点遍历形式，返回 TimeValuePair （主要用于带值过滤的查询） -* 按递增时间戳查询对应值（主要用于带值过滤的查询） +* 按递增时间戳查询对应值的接口（主要用于带值过滤的查询） ## 相关接口 -以上四种读取单个时间序列数据的方式对应代码里的四个接口 - -### org.apache.iotdb.tsfile.read.reader.IPointReader - -#### 主要方法 - -``` -// 判断是否还有数据点 -boolean hasNextTimeValuePair() throws IOException; - -// 获得当前数据点，并把游标后移 -TimeValuePair nextTimeValuePair() throws IOException; - -// 获得当前数据点，不移动游标 -TimeValuePair currentTimeValuePair() throws IOException; -``` - -#### 一般使用流程 - -``` -while (pointReader.hasNextTimeValuePair()) { - TimeValuePair timeValuePair = pointReader.currentTimeValuePair(); - - // use current timeValuePair to do some work - ... - - // get next timeValuePair - timeValuePair = mergeReader.nextTimeValuePair(); -} -``` +以上三种读取单个时间序列数据的方式对应代码里的三个接口 ### org.apache.iotdb.tsfile.read.reader.IBatchReader #### 主要方法 ``` -// 判断是否还有BatchData +// 判断是否还有 BatchData boolean hasNextBatch() throws IOException; -// 获得下一个BatchData，并把游标后移 +// 获得下一个 BatchData，并把游标后移 BatchData nextBatch() throws IOException; ``` -#### 一般使用流程 +#### 使用流程 ``` while (batchReader.hasNextBatch()) { @@ -86,60 +56,37 @@ while (batchReader.hasNextBatch()) { } ``` -### org.apache.iotdb.db.query.reader.IReaderByTimestamp +### org.apache.iotdb.db.query.reader.series.IAggregateReader #### 主要方法 -``` -// 得到给定时间戳的值，如果不存在返回null（要求传入的 timestamp 是递增的） -Object getValueInTimestamp(long timestamp) throws IOException; -``` - -#### 一般使用流程 - -该接口在带值过滤的查询中被使用，TimeGenerator生成时间戳后，使用该接口获得该时间戳对应的value - ``` -while (timeGenerator.hasNext()) { - long timestamp = timeGenerator.next(); - Object value = readerByTimestamp.getValueInTimestamp(timestamp); -} -``` - -### org.apache.iotdb.db.query.reader.seriesRelated.IAggregateReader - -#### 主要方法 - -``` -// 判断是否还有Chunk +// 判断是否还有 Chunk boolean hasNextChunk() throws IOException; -// 判断是否能够使用当前Chunk的统计信息 +// 判断是否能够使用当前 Chunk 的统计信息 boolean canUseCurrentChunkStatistics(); -// 获得当前Chunk的统计信息 +// 获得当前 Chunk 的统计信息 Statistics currentChunkStatistics(); -// 跳过当前Chunk +// 跳过当前 Chunk void skipCurrentChunk(); -// 判断当前Chunk是否还有下一个Page +// 判断当前Chunk是否还有下一个 Page boolean hasNextPage() throws IOException; -// 判断能否使用当前Page的统计信息 +// 判断能否使用当前 Page 的统计信息 boolean canUseCurrentPageStatistics() throws IOException; -// 获得当前Page的统计信息 +// 获得当前 Page 的统计信息 Statistics currentPageStatistics() throws IOException; -// 跳过当前的Page +// 跳过当前的 Page void skipCurrentPage(); -// 判断是否有下一个重叠的Page -boolean hasNextOverlappedPage() throws IOException; - -// 获得当前重叠Page的数据 -BatchData nextOverlappedPage() throws IOException; +// 获得当前 Page 的数据 +BatchData nextPage() throws IOException; ``` #### 一般使用流程 @@ -149,266 +96,289 @@ while (aggregateReader.hasNextChunk()) { if (aggregateReader.canUseCurrentChunkStatistics()) { Statistics chunkStatistics = aggregateReader.currentChunkStatistics(); - // do some aggregate calculation using chunk statistics + // 用 chunk 层的统计信息计算 ... aggregateReader.skipCurrentChunk(); continue; } - + + // 把当前 chunk 中的 page 消耗完 while (aggregateReader.hasNextPage()) { if (aggregateReader.canUseCurrentPageStatistics()) { + // 可以用统计信息 Statistics pageStatistic = aggregateReader.currentPageStatistics(); - // do some aggregate calculation using page statistics - ... - + // 用 page 层的统计信息计算 + ... + aggregateReader.skipCurrentPage(); continue; - } - - // 遍历所有重叠的page - while (aggregateReader.hasNextOverlappedPage()) { + } else { + // 不能用统计信息，需要用数据计算 BatchData batchData = aggregateReader.nextOverlappedPage(); - // do some aggregate calculation using batch data - ... + // 用 batchData 计算 + ... } } } ``` -## 具体实现类 - -上述四个接口都有其对应的实现类，这四个实现类的内部都有一个 SeriesReader 对象，SeriesReader是一个基础的工具类， -封装了对于一个时间序列读取操作的基本方法。所以，下面先介绍一下SeriesReader的实现，然后再依次介绍四个接口实现类的具体实现。 +### org.apache.iotdb.db.query.reader.IReaderByTimestamp -### org.apache.iotdb.db.query.reader.seriesRelated.SeriesReader +#### 主要方法 -首先介绍一下SeriesReader里的几个重要字段 +``` +// 得到给定时间戳的值，如果不存在返回 null（要求传入的 timestamp 是递增的） +Object getValueInTimestamp(long timestamp) throws IOException; -* private final List<TsFileResource> seqFileResource; +// 给定一批递增时间戳的值，返回一批结果（减少方法调用次数） +Object[] getValuesInTimestamps(long[] timestamps) throws IOException; +``` - 顺序文件列表，因为顺序文件本身就保证有序，且时间戳互不重叠，只需使用List进行存储 - -* private final PriorityQueue<TsFileResource> unseqFileResource; - - 乱序文件列表，因为乱序文件互相之间不保证顺序性，且可能有重叠，为了保证顺序，使用优先队列进行存储 +#### 一般使用流程 -* private final List<ChunkMetaData> seqChunkMetadatas = new LinkedList<>(); +该接口在带值过滤的查询中被使用，TimeGenerator生成时间戳后，使用该接口获得该时间戳对应的value - 顺序文件解开后得到的ChunkMetaData列表，本身有序且互不重叠，所以使用List存储 - -* private final PriorityQueue<ChunkMetaData> unseqChunkMetadatas = - new PriorityQueue<>(Comparator.comparingLong(ChunkMetaData::getStartTime)); - - 乱序文件解开后得到的ChunkMetaData列表，互相之间可能有重叠，为了保证顺序，使用优先队列进行存储 +``` +Object value = readerByTimestamp.getValueInTimestamp(timestamp); -* private boolean hasCachedFirstChunkMetadata; +or - 是否缓存了第一个chunk meta data - -* private ChunkMetaData firstChunkMetaData; +Object[] values = readerByTimestamp.getValueInTimestamp(timestamps); +``` - 当前第一个chunk meta data的引用 +## 具体实现类 -* private PriorityQueue<VersionPair<IPageReader>> cachedPageReaders = - new PriorityQueue<>( - Comparator.comparingLong(pageReader -> pageReader.data.getStatistics().getStartTime())); - - 所有重叠page的reader，按照每个page的起始时间进行排序 - -* private PriorityMergeReader mergeReader = new PriorityMergeReader(); +上述三个接口都有其对应的实现类，由于以上三种查询有共性，我们设计了一个基础的 SeriesReader 工具类，封装了对于一个时间序列读取操作的基本方法，帮助实现以上三种接口。下面首先介绍 SeriesReader 的设计原理，然后再依次介绍三个接口的具体实现。 - PriorityMergeReader可以接收多个数据源，且根据其优先级输出优先级最高的数据点 +### org.apache.iotdb.db.query.reader.series.SeriesReader -* private boolean hasCachedNextBatch; +#### 设计思想 - 是否缓存了下一个batch - -* private BatchData cachedBatchData; +背景知识：TsFile 文件（TsFileResource）解开后可以得到 ChunkMetadata，ChunkMetadata 解开后可以得到一堆 PageReader，PageReader 可以直接返回 BatchData 数据点。 - 缓存的下一个batch的引用 - -下面介绍一下SeriesReader里的重要方法 +为了支持以上三种接口 -#### hasNextChunk() +数据按照粒度从大到小分成四种：文件，Chunk，Page，相交数据点。在原始数据查询中，最大的数据块返回粒度是一个 page，如果一个 page 和其他 page 由于乱序写入相互覆盖了，就解开成数据点做合并。聚合查询中优先使用 Chunk 的统计信息，其次是 Page 的统计信息，最后是相交数据点。 -这个方法判断该时间序列还有没有下一个chunk。 +设计原则是能用粒度大的就不用粒度小的。 -如果`hasCachedFirstChunkMetadata`为`true`，则代表当前已经缓存了第一个`ChunkMetaData`，且该`ChunkMetaData`并未被使用，直接返回`true`；若`hasCachedFirstChunkMetadata`为`false`，我们调用`tryToInitFirstChunk()`，尝试去解开顺序文件和乱序文件，并从中选出开始时间最小的chunk meta data赋值给`firstChunkMetaData`，并将`hasCachedFirstChunkMetadata`置为`true`。 +首先介绍一下SeriesReader里的几个重要字段 -#### tryToInitFirstChunk() +``` -这个方法尝试去初始化当前的chunk meta data。 +/* + * 文件层 + */ +private final List<TsFileResource> seqFileResource; + 顺序文件列表，因为顺序文件本身就保证有序，且时间戳互不重叠，只需使用List进行存储 + +private final PriorityQueue<TsFileResource> unseqFileResource; + 乱序文件列表，因为乱序文件互相之间不保证顺序性，且可能有重叠，为了保证顺序，使用优先队列进行存储 + +/* + * chunk 层 + * + * 三个字段之间数据永远不重复，first 永远是第一个（开始时间最小） + */ +private ChunkMetaData firstChunkMetaData; + 填充 chunk 层时优先填充此字段，保证这个 chunk 具有当前最小开始时间 + +private final List<ChunkMetaData> seqChunkMetadatas; + 顺序文件解开后得到的 ChunkMetaData 存放在此，本身有序且互不重叠，所以使用 List 存储 -首先调用`tryToFillChunkMetadatas()`去尝试解开顺序和乱序文件，并填充`seqChunkMetadatas`和`unseqChunkMetadatas`，之后再从这二者里选出开始时间最小的。如果只有顺序的，直接取出`seqChunkMetadatas`的第一个；如果只有乱序的，直接取出`unseqChunkMetadatas`的第一个，如果既有顺序的又有乱序的，比较两者第一个的开始时间大小，取较小者取出。 +private final PriorityQueue<ChunkMetaData> unseqChunkMetadatas; + 乱序文件解开后得到的 ChunkMetaData 存放在此，互相之间可能有重叠，为了保证顺序，使用优先队列进行存储 + +/* + * page 层 + * + * 两个字段之间数据永远不重复，first 永远是第一个（开始时间最小） + */ +private VersionPageReader firstPageReader; + 开始时间最小的 page reader + +private PriorityQueue<VersionPageReader> cachedPageReaders; + 当前获得的所有 page reader，按照每个 page 的起始时间进行排序 + +/* + * 相交数据点层 + */ +private PriorityMergeReader mergeReader; + 本质上是多个带优先级的 page，按时间戳从低到高输出数据点，时间戳相同时，保留优先级高的 + +/* + * 相交数据点产出结果的缓存 + */ +private boolean hasCachedNextOverlappedPage; + 是否缓存了下一个batch + +private BatchData cachedBatchData; + 缓存的下一个batch的引用 +``` + +下面介绍一下SeriesReader里的重要方法 -#### tryToFillChunkMetadatas() +#### hasNextChunk() -这个方法尝试解开顺序和乱序文件。 +* 主要功能：判断该时间序列还有没有下一个chunk。 -因为可能一个时间序列对应很多个文件，所以一下子解开全部的文件，可能会导致OOM，所以我们推迟解开文件的时间至其被需要时。 +* 约束：在调用这个方法前，需要保证 `SeriesReader` 内已经没有 page 和 数据点 层级的数据了，也就是之前解开的 chunk 都消耗完了。 -因为顺序文件之间相互不重叠，所以，对于顺序文件，我们只需要保证`seqChunkMetadatas`不为空即可，每次检测到`seqChunkMetadatas`为空后，调用`loadSatisfiedChunkMetadatas(TsFileResource)`方法去解开`seqFileResource`的第一个顺序文件。 +* 实现：如果 `firstChunkMetaData` 不为空，则代表当前已经缓存了第一个 `ChunkMetaData`，且未被使用，直接返回`true`； -而乱序文件之间可能会相互重叠，仅仅解开第一个乱序文件是不够的，还需要解开与`unseqChunkMetadatas`中第一个chunk meta data时间有重叠的所有乱序文件。 + 尝试去解开第一个顺序文件和第一个乱序文件，填充 chunk 层。并解开与 `firstChunkMetadata` 相重合的所有文件。 #### isChunkOverlapped() -这个方法判断当前的chunk有没有其他与之重叠的chunk存在。 +* 主要功能：判断当前的 chunk 有没有与其他 Chunk 有重叠 -如果`mergeReader`里仍然有数据，或者`seqChunkMetadatas`里有与`firstChunkMetaData`时间重叠的，或者`unseqChunkMetadatas`里有与`firstChunkMetaData`时间重叠的，则返回`true`；反之，返回`false`。 +* 约束：在调用这个方法前，需要保证 chunk 层已经缓存了 `firstChunkMetadata`，也就是调用了 hasNextChunk() 并且为 true。 + +* 实现：直接把 `firstChunkMetadata` 与 `seqChunkMetadatas` 和 `unseqChunkMetadatas` 相比较。因为此前已经保证所有和 `firstChunkMetadata` 相交的文件都被解开了。 #### currentChunkStatistics() -返回`firstChunkMetaData`的统计信息。 +返回 `firstChunkMetaData` 的统计信息。 #### skipCurrentChunk() -跳过当前chunk。只需要将`hasCachedFirstChunkMetadata`置为`false`，`firstChunkMetaData`置为`null`即可。 +跳过当前 chunk。只需要将`firstChunkMetaData `置为`null`。 #### hasNextPage() -这个方法判断是否有下一个Page，一般在`firstChunkMetaData`不可直接使用时，继续解成Page。 - -首先调用`fillOverlappedPageReaders()`去将`firstChunkMetaData`解开为`PageReader`，解开的`PageReader`都放进`cachedPageReaders`里。并将`hasCachedFirstChunkMetadata`置为`false`，`firstChunkMetaData`置为`null`。若`cachedPageReaders`为空则返回`false`，若不为空，返回`true`。 +* 主要功能：判断 SeriesReader 中还有没有已经解开的 page，如果有相交的 page，就构造 `cachedBatchData` 并缓存，否则缓存 `firstPageReader`。 -#### isPageOverlapped() +* 实现：如果已经缓存了 `cachedBatchData` 就直接返回。如果有相交数据点，就构造 `cachedBatchData`。如果已经缓存了 `firstPageReader`，就直接返回。 -这个方法判断当前的Page有没有其他与之重叠的Page存在。 + 如果当前的 `firstChunkMetadata` 还没有解开，就解开与之重叠的所有 ChunkMetadata，构造 firstPageReader。 + + 判断，如果 `firstPageReader` 和 `cachedPageReaders` 相交，则构造 `cachedBatchData`，否则直接返回。 -如果`mergeReader`里仍然有数据，或者`seqChunkMetadatas`里有与`cachedPageReaders`里第一个`PageReader`时间重叠的，或者`unseqChunkMetadatas`里有与`cachedPageReaders`里第一个`PageReader`时间重叠的，则返回`true`；反之，返回`false`。 +#### isPageOverlapped() -#### nextPage() +* 主要功能：判断当前的 page 有没有与其他 page 有重叠 -须与`isPageOverlapped()`方法搭配使用。 +* 约束：在调用这个方法前，需要保证调用了 hasNextPage() 并且为 true。也就是，有可能缓存了一个相交的 `cachedBatchData`，或者缓存了不相交的 `firstPageReader`。 -当`cachedPageReaders`里第一个Page没有与之重叠的其他Page时，直接获得`cachedPageReaders`的第一个Page里符合过滤条件的所有data。 +* 实现：先判断有没有 `cachedBatchData`，如果没有，就说明当前 page 不相交，则 `mergeReader` 里没数据。再判断 `firstPageReader` 是否与 `cachedPageReaders` 中的 page 相交。 #### currentPageStatistics() -返回`cachedPageReaders`里第一个Page的统计信息。 +返回 `firstPageReader` 的统计信息。 #### skipCurrentPage() -跳过当前Page。只需要将`cachedPageReaders`里第一个PageReader删掉即可。 +跳过当前Page。只需要将 `firstPageReader` 置为 null。 -#### hasNextOverlappedPage() +#### nextPage() -这个方法判断当前还有没有重叠的Page。 +* 主要功能：返回下一个相交或不想交的 page -如果`hasCachedNextBatch`为`true`，直接返回`true`。 +* 约束：在调用这个方法前，需要保证调用了 hasNextPage() 并且为 true。也就是，有可能缓存了一个相交的 `cachedBatchData`，或者缓存了不相交的 `firstPageReader`。 -否则，先调用`putAllDirectlyOverlappedPageReadersIntoMergeReader()`方法，将所有与`cachedPageReaders`第一个Page有重叠的PageReader放进`mergeReader`里。`mergeReader`里维护了一个`currentLargestEndTime`变量，每次add进新的Reader时被更新，用以记录当前添加进`mergeReader`的最大的结束时间。 +* 实现：如果 `hasCachedNextOverlappedPage` 为 true，说明缓存了一个相交的 page，直接返回 `cachedBatchData`。否则当前 page 不相交，直接从 firstPageReader 里拿当前 page 的数据。 -然后先从`mergeReader`里取出当前最大的结束时间，作为此次所要返回的batch的结束时间，记为`currentPageEndTime`。接着去遍历`mergeReader`，直到当前的时间戳大于`currentPageEndTime`。 +#### hasNextOverlappedPage() -在迭代的过程里，我们也要判断是否有与当前时间戳重叠的file或者chunk或者page（这里之所以还要再做一次判断，是因为，比如当前page是1-30，和他直接相交的page是20-50，还有一个page是40-60，每取一个点判断一次是想把40-60解开），如果有，解开相应的file或者chunk或者page，并将其放入`mergeReader`。完成重叠的判断后，从`mergeReader`中取出相应数据。 +* 主要功能：内部方法，用来判断当前有没有重叠的数据，并且构造相交的 page 并缓存。 -完成迭代后将获得数据缓存在`cachedBatchData`中，并将`hasCachedNextBatch`置为`true`。 +* 实现：如果 `hasCachedNextOverlappedPage` 为 `true`，直接返回 `true`。 -#### nextOverlappedPage() + 否则，先调用 `tryToPutAllDirectlyOverlappedPageReadersIntoMergeReader()` 方法，将 `cachedPageReaders` 中所有与 `firstPageReader` 有重叠的放进 `mergeReader` 里。`mergeReader` 里维护了一个 `currentLargestEndTime` 变量，每次添加进新的 Reader 时被更新，用以记录当前添加进 `mergeReader` 的 page 的最大结束时间。 + 然后先从`mergeReader`里取出当前最大的结束时间，作为第一批数据的结束时间，记为`currentPageEndTime`。接着去遍历`mergeReader`，直到当前的时间戳大于`currentPageEndTime`。 + + 每从 mergeReader 移出一个点前，我们先要判断是否有与当前时间戳重叠的file或者chunk或者page（这里之所以还要再做一次判断，是因为，比如当前page是1-30，和他直接相交的page是20-50，还有一个page是40-60，每取一个点判断一次是想把40-60解开），如果有，解开相应的file或者chunk或者page，并将其放入`mergeReader`。完成重叠的判断后，从`mergeReader`中取出相应数据。 -将缓存的`cachedBatchData`返回，并将`hasCachedNextBatch`置为`false`。 + 完成迭代后将获得数据缓存在 `cachedBatchData` 中，并将 `hasCachedNextOverlappedPage` 置为 `true`。 -### org.apache.iotdb.db.query.reader.seriesRelated.SeriesRawDataPointReader +#### nextOverlappedPage() -`SeriesRawDataPointReader`实现了`IPointReader`。 +将缓存的`cachedBatchData`返回，并将`hasCachedNextOverlappedPage`置为`false`。 -其方法`hasNextTimeValuePair()`的核心判断流程是去组合调用SeriesReader的三个方法 - -``` -while (seriesReader.hasNextChunk()) { - while (seriesReader.hasNextPage()) { - if (seriesReader.hasNextOverlappedPage()) { - return true; - } - } -} -return false; -``` - -### org.apache.iotdb.db.query.reader.seriesRelated.SeriesRawDataBatchReader +### org.apache.iotdb.db.query.reader.series.SeriesRawDataBatchReader `SeriesRawDataBatchReader`实现了`IBatchReader`。 其方法`hasNextBatch()`的核心判断流程是 ``` -// 判断有没有下一个chunk +// 有缓存了 batch，直接返回 +if (hasCachedBatchData) { + return true; +} + +/* + * 如果 SeriesReader 里还有 page，返回 page + */ +if (readPageData()) { + hasCachedBatchData = true; + return true; +} + +/* + * 如果有 chunk，并且有 page，返回 page + */ while (seriesReader.hasNextChunk()) { - // 判断有没有下一个page - while (seriesReader.hasNextPage()) { - // 如果当前page没有重叠 - if (!seriesReader.isPageOverlapped()) { - // 直接调用nextPage()方法返回一个batch - batchData = seriesReader.nextPage(); - hasCachedBatchData = true; - return true; - } - // 如果当前还有下一个重叠的Page - if (seriesReader.hasNextOverlappedPage()) { - // 调用nextOverlappedPage()方法返回一个batch - batchData = seriesReader.nextOverlappedPage(); - hasCachedBatchData = true; - return true; - } + if (readPageData()) { + hasCachedBatchData = true; + return true; } } -return false; +return hasCachedBatchData; ``` -### org.apache.iotdb.db.query.reader.seriesRelated.SeriesReaderByTimestamp +### org.apache.iotdb.db.query.reader.series.SeriesReaderByTimestamp + +`SeriesReaderByTimestamp` 实现了 `IReaderByTimestamp`。 -`SeriesReaderByTimestamp`实现了`IReaderByTimestamp`。 +设计思想：当给一个时间戳要查询值时，这个时间戳可以转化为一个 time >= x 的过滤条件。不断更新这个过滤条件，并且跳过不满足的文件，chunk 和 page。 -其方法getValueInTimestamp()的核心流程是找到下一个满足条件的batch data，然后调`BatchData`的`getValueInTimestamp(long)`方法获得相应值，而找下一个batch data的流程是 +实现方式： ``` -// 判断是否有下一个chunk +/* + * 优先判断下一个 page 有没有当前所查时间，能跳过就跳过 + */ +if (readPageData(timestamp)) { + return true; +} + +/* + * 判断下一个 chunk 有没有当前所查时间，能跳过就跳过 + */ while (seriesReader.hasNextChunk()) { - // 如果当前chunk不满足过滤条件，直接跳过当前chunk，进入下一个循环 - if (!satisfyTimeFilter(seriesReader.currentChunkStatistics())) { + Statistics statistics = seriesReader.currentChunkStatistics(); + if (!satisfyTimeFilter(statistics)) { seriesReader.skipCurrentChunk(); continue; } - // 判断是否有下一个page - while (seriesReader.hasNextPage()) { - // 如果当前page不满足过滤条件，直接跳过当前page，进入下一个循环 - if (!satisfyTimeFilter(seriesReader.currentPageStatistics())) { - seriesReader.skipCurrentPage(); - continue; - } - // 如果当前的page没有重叠 - if (!seriesReader.isPageOverlapped()) { - // 直接调用nextPage()方法获得下一个batch - batchData = seriesReader.nextPage(); - } - // 如果当前的page有重叠 - else { - // 调用nextOverlappedPage()方法获得下一个batch - batchData = seriesReader.nextOverlappedPage(); - } - // 如果该batch满足要求，则返回true，否则继续搜索下一个batch - if (batchData.getTimeByIndex(batchData.length() - 1) >= timestamp) { - return true; - } + /* + * chunk 不能跳过，继续到 chunk 里检查 page + */ + if (readPageData(timestamp)) { + return true; } } return false; ``` -### org.apache.iotdb.db.query.reader.seriesRelated.SeriesAggregateReader +### org.apache.iotdb.db.query.reader.series.SeriesAggregateReader -`SeriesAggregateReader`实现了`IAggregateReader` +`SeriesAggregateReader` 实现了 `IAggregateReader` `IAggregateReader`的大部分接口方法都在`SeriesReader`有对应实现，除了`canUseCurrentChunkStatistics()`和`canUseCurrentPageStatistics()`两个方法。 #### canUseCurrentChunkStatistics() +设计思想：可以用统计信息的条件是当前 chunk 不重叠，并且满足过滤条件。 + 先调用`SeriesReader`的`currentChunkStatistics()`方法，获得当前chunk的统计信息，再调用`SeriesReader`的`isChunkOverlapped()`方法判断当前chunk是否重叠，如果当前chunk不重叠，且其统计信息满足过滤条件，则返回`true`，否则返回`false`。 #### canUseCurrentPageStatistics() -先调用`SeriesReader`的`currentPageStatistics()`方法，获得当前page的统计信息，再调用`SeriesReader`的`isPageOverlapped()`方法判断当前page是否重叠，如果当前page不重叠，且其统计信息满足过滤条件，则返回`true`，否则返回`false`。 +设计思想：可以用统计信息的条件是当前 page 不重叠，并且满足过滤条件。 + +先调用`SeriesReader`的 `currentPageStatistics()` 方法，获得当前page的统计信息，再调用`SeriesReader` 的 `isPageOverlapped()` 方法判断当前 page 是否重叠，如果当前page不重叠，且其统计信息满足过滤条件，则返回`true`，否则返回`false`。 \ No newline at end of file diff --git a/docs/Documentation-CHN/SystemDesign/5-DataQuery/4-AggregationQuery.md b/docs/Documentation-CHN/SystemDesign/5-DataQuery/4-AggregationQuery.md index 31f3c4a..56b2589 100644 --- a/docs/Documentation-CHN/SystemDesign/5-DataQuery/4-AggregationQuery.md +++ b/docs/Documentation-CHN/SystemDesign/5-DataQuery/4-AggregationQuery.md @@ -63,15 +63,11 @@ while (aggregateReader.hasNextChunk()) { aggregateReader.skipCurrentPage(); continue; - } - - // 遍历所有重叠的page - while (aggregateReader.hasNextOverlappedPage()) { - BatchData batchData = aggregateReader.nextOverlappedPage(); - - // do some aggregate calculation using batch data + } else { + BatchData batchData = aggregateReader.nextPage(); + // do some aggregate calculation using batch data ... - } + } } } ``` diff --git a/docs/Documentation-CHN/SystemDesign/5-DataQuery/5-GroupByQuery.md b/docs/Documentation-CHN/SystemDesign/5-DataQuery/5-GroupByQuery.md index 6fa7d22..911ee7e 100644 --- a/docs/Documentation-CHN/SystemDesign/5-DataQuery/5-GroupByQuery.md +++ b/docs/Documentation-CHN/SystemDesign/5-DataQuery/5-GroupByQuery.md @@ -89,6 +89,7 @@ if (aggregateResultList.length == 0) { ``` while (aggregateReader.hasNextChunk()) { if (aggregateReader.canUseCurrentChunkStatistics()) { + // 可以用 chunk 统计信息 Statistics chunkStatistics = aggregateReader.currentChunkStatistics(); // do some aggregate calculation using chunk statistics @@ -97,23 +98,19 @@ while (aggregateReader.hasNextChunk()) { aggregateReader.skipCurrentChunk(); continue; } - + + // chunk 统计信息不能用，消耗所有 page while (aggregateReader.hasNextPage()) { if (aggregateReader.canUseCurrentPageStatistics()) { + // 可以用 page 统计信息 Statistics pageStatistic = aggregateReader.currentPageStatistics(); - - // do some aggregate calculation using page statistics ... aggregateReader.skipCurrentPage(); continue; - } - - // 遍历所有重叠的page - while (aggregateReader.hasNextOverlappedPage()) { - BatchData batchData = aggregateReader.nextOverlappedPage(); - - // do some aggregate calculation using batch data + } else { + // page 统计信息不能用，用数据计算 + BatchData batchData = aggregateReader.nextPage(); ... } } diff --git a/server/src/main/java/org/apache/iotdb/db/query/reader/series/IAggregateReader.java b/server/src/main/java/org/apache/iotdb/db/query/reader/series/IAggregateReader.java index 23e5dde..4893970 100644 --- a/server/src/main/java/org/apache/iotdb/db/query/reader/series/IAggregateReader.java +++ b/server/src/main/java/org/apache/iotdb/db/query/reader/series/IAggregateReader.java @@ -27,7 +27,7 @@ public interface IAggregateReader { boolean hasNextChunk() throws IOException; - boolean canUseCurrentChunkStatistics(); + boolean canUseCurrentChunkStatistics() throws IOException; Statistics currentChunkStatistics(); diff --git a/server/src/main/java/org/apache/iotdb/db/query/reader/series/SeriesAggregateReader.java b/server/src/main/java/org/apache/iotdb/db/query/reader/series/SeriesAggregateReader.java index 453d6af..7a6e40b 100644 --- a/server/src/main/java/org/apache/iotdb/db/query/reader/series/SeriesAggregateReader.java +++ b/server/src/main/java/org/apache/iotdb/db/query/reader/series/SeriesAggregateReader.java @@ -46,7 +46,7 @@ public class SeriesAggregateReader implements IAggregateReader { } @Override - public boolean canUseCurrentChunkStatistics() { + public boolean canUseCurrentChunkStatistics() throws IOException { Statistics chunkStatistics = currentChunkStatistics(); return !seriesReader.isChunkOverlapped() && containedByTimeFilter(chunkStatistics); } @@ -77,7 +77,7 @@ public class SeriesAggregateReader implements IAggregateReader { } @Override - public Statistics currentPageStatistics() throws IOException { + public Statistics currentPageStatistics() { return seriesReader.currentPageStatistics(); } diff --git a/server/src/main/java/org/apache/iotdb/db/query/reader/series/SeriesRawDataPointReader.java b/server/src/main/java/org/apache/iotdb/db/query/reader/series/SeriesRawDataPointReader.java index dd3296c..7885b94 100644 --- a/server/src/main/java/org/apache/iotdb/db/query/reader/series/SeriesRawDataPointReader.java +++ b/server/src/main/java/org/apache/iotdb/db/query/reader/series/SeriesRawDataPointReader.java @@ -24,6 +24,9 @@ import org.apache.iotdb.tsfile.read.reader.IPointReader; import java.io.IOException; +/** + * only for test now + */ public class SeriesRawDataPointReader implements IPointReader { private final SeriesRawDataBatchReader batchReader; diff --git a/server/src/main/java/org/apache/iotdb/db/query/reader/series/SeriesReader.java b/server/src/main/java/org/apache/iotdb/db/query/reader/series/SeriesReader.java index 2c0765a..1d7825d 100644 --- a/server/src/main/java/org/apache/iotdb/db/query/reader/series/SeriesReader.java +++ b/server/src/main/java/org/apache/iotdb/db/query/reader/series/SeriesReader.java @@ -18,25 +18,20 @@ */ package org.apache.iotdb.db.query.reader.series; -import org.apache.iotdb.db.engine.cache.DeviceMetaDataCache; -import org.apache.iotdb.db.engine.modification.Modification; import org.apache.iotdb.db.engine.querycontext.QueryDataSource; -import org.apache.iotdb.db.engine.querycontext.ReadOnlyMemChunk; import org.apache.iotdb.db.engine.storagegroup.TsFileResource; import org.apache.iotdb.db.query.context.QueryContext; -import org.apache.iotdb.db.query.control.FileReaderManager; import org.apache.iotdb.db.query.filter.TsFileFilter; -import org.apache.iotdb.db.query.reader.chunk.DiskChunkLoader; import org.apache.iotdb.db.query.reader.chunk.MemChunkLoader; import org.apache.iotdb.db.query.reader.chunk.MemChunkReader; import org.apache.iotdb.db.query.reader.universal.PriorityMergeReader; +import org.apache.iotdb.db.utils.FileLoaderUtils; import org.apache.iotdb.db.utils.QueryUtils; import org.apache.iotdb.db.utils.TestOnly; 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.Statistics; import org.apache.iotdb.tsfile.read.TimeValuePair; -import org.apache.iotdb.tsfile.read.TsFileSequenceReader; import org.apache.iotdb.tsfile.read.common.BatchData; import org.apache.iotdb.tsfile.read.common.Chunk; import org.apache.iotdb.tsfile.read.common.Path; @@ -143,7 +138,11 @@ class SeriesReader { } - boolean isChunkOverlapped() { + boolean isChunkOverlapped() throws IOException { + if (firstChunkMetaData == null) { + throw new IOException("no first chunk"); + } + Statistics chunkStatistics = firstChunkMetaData.getStatistics(); return !seqChunkMetadatas.isEmpty() && chunkStatistics.getEndTime() >= seqChunkMetadatas.get(0).getStartTime() @@ -276,6 +275,17 @@ class SeriesReader { firstPageStatistics.getEndTime() >= cachedPageReaders.peek().getStartTime(); } + Statistics currentPageStatistics() { + if (firstPageReader == null) { + return null; + } + return firstPageReader.getStatistics(); + } + + void skipCurrentPage() { + firstPageReader = null; + } + /** * This method should only be used when the method isPageOverlapped() return true. */ @@ -303,17 +313,6 @@ class SeriesReader { } } - Statistics currentPageStatistics() { - if (firstPageReader == null) { - return null; - } - return firstPageReader.getStatistics(); - } - - void skipCurrentPage() { - firstPageReader = null; - } - /** * read overlapped data till currentLargestEndTime in mergeReader, * if current batch does not contain data, read till next currentLargestEndTime again @@ -445,46 +444,6 @@ class SeriesReader { return chunkReader; } - private List<ChunkMetaData> loadSatisfiedChunkMetadatas(TsFileResource resource) - throws IOException { - List<ChunkMetaData> currentChunkMetaDataList; - if (resource == null) { - return new ArrayList<>(); - } - if (resource.isClosed()) { - currentChunkMetaDataList = DeviceMetaDataCache.getInstance().get(resource, seriesPath); - } else { - currentChunkMetaDataList = resource.getChunkMetaDataList(); - } - List<Modification> pathModifications = - context.getPathModifications(resource.getModFile(), seriesPath.getFullPath()); - - if (!pathModifications.isEmpty()) { - QueryUtils.modifyChunkMetaData(currentChunkMetaDataList, pathModifications); - } - - for (ChunkMetaData chunkMetaData : currentChunkMetaDataList) { - TsFileSequenceReader tsFileSequenceReader = FileReaderManager.getInstance() - .get(resource, resource.isClosed()); - chunkMetaData.setChunkLoader(new DiskChunkLoader(tsFileSequenceReader)); - } - List<ReadOnlyMemChunk> memChunks = resource.getReadOnlyMemChunk(); - if (memChunks != null) { - for (ReadOnlyMemChunk readOnlyMemChunk : memChunks) { - if (!memChunks.isEmpty()) { - currentChunkMetaDataList.add(readOnlyMemChunk.getChunkMetaData()); - } - } - } - - /* - * remove empty and not satisfied ChunkMetaData - */ - currentChunkMetaDataList.removeIf(chunkMetaData -> (timeFilter != null && !timeFilter - .satisfyStartEndTime(chunkMetaData.getStartTime(), chunkMetaData.getEndTime())) - || chunkMetaData.getStartTime() > chunkMetaData.getEndTime()); - return currentChunkMetaDataList; - } private PriorityQueue<TsFileResource> sortUnSeqFileResources( List<TsFileResource> tsFileResources) { @@ -516,14 +475,18 @@ class SeriesReader { * Fill sequence chunkMetadatas until it is not empty */ while (seqChunkMetadatas.isEmpty() && !seqFileResource.isEmpty()) { - seqChunkMetadatas.addAll(loadSatisfiedChunkMetadatas(seqFileResource.remove(0))); + seqChunkMetadatas.addAll(FileLoaderUtils + .loadChunkMetadataFromTsFileResource(seqFileResource.remove(0), seriesPath, context, + timeFilter)); } /* * Fill unsequence chunkMetadatas until it is not empty */ while (unseqChunkMetadatas.isEmpty() && !unseqFileResource.isEmpty()) { - unseqChunkMetadatas.addAll(loadSatisfiedChunkMetadatas(unseqFileResource.poll())); + unseqChunkMetadatas.addAll(FileLoaderUtils + .loadChunkMetadataFromTsFileResource(unseqFileResource.poll(), seriesPath, context, + timeFilter)); } /* @@ -555,11 +518,15 @@ class SeriesReader { private void unpackAllOverlappedTsFilesToChunkMetadatas(long endTime) throws IOException { while (!unseqFileResource.isEmpty() && endTime >= unseqFileResource.peek().getStartTimeMap().get(seriesPath.getDevice())) { - unseqChunkMetadatas.addAll(loadSatisfiedChunkMetadatas(unseqFileResource.poll())); + unseqChunkMetadatas.addAll(FileLoaderUtils + .loadChunkMetadataFromTsFileResource(unseqFileResource.poll(), seriesPath, context, + timeFilter)); } while (!seqFileResource.isEmpty() && endTime >= seqFileResource.get(0).getStartTimeMap().get(seriesPath.getDevice())) { - seqChunkMetadatas.addAll(loadSatisfiedChunkMetadatas(seqFileResource.remove(0))); + seqChunkMetadatas.addAll(FileLoaderUtils + .loadChunkMetadataFromTsFileResource(seqFileResource.remove(0), seriesPath, context, + timeFilter)); } } diff --git a/server/src/main/java/org/apache/iotdb/db/utils/FileLoaderUtils.java b/server/src/main/java/org/apache/iotdb/db/utils/FileLoaderUtils.java index 671e5d4..7b93c94 100644 --- a/server/src/main/java/org/apache/iotdb/db/utils/FileLoaderUtils.java +++ b/server/src/main/java/org/apache/iotdb/db/utils/FileLoaderUtils.java @@ -35,6 +35,7 @@ import org.apache.iotdb.tsfile.file.metadata.TsDeviceMetadataIndex; import org.apache.iotdb.tsfile.file.metadata.TsFileMetaData; import org.apache.iotdb.tsfile.read.TsFileSequenceReader; import org.apache.iotdb.tsfile.read.common.Path; +import org.apache.iotdb.tsfile.read.filter.basic.Filter; public class FileLoaderUtils { @@ -73,25 +74,44 @@ public class FileLoaderUtils { } } + /** + * load all ChunkMetadatas belong to the seriesPath and satisfy filter + */ + public static List<ChunkMetaData> loadChunkMetadataFromTsFileResource( + TsFileResource resource, Path seriesPath, QueryContext context, Filter timeFilter) throws IOException { + List<ChunkMetaData> chunkMetaDataList = loadChunkMetadataFromTsFileResource(resource, seriesPath, context); + + /* + * remove not satisfied ChunkMetaData + */ + chunkMetaDataList.removeIf(chunkMetaData -> (timeFilter != null && !timeFilter + .satisfyStartEndTime(chunkMetaData.getStartTime(), chunkMetaData.getEndTime())) + || chunkMetaData.getStartTime() > chunkMetaData.getEndTime()); + return chunkMetaDataList; + } + + /** + * load all ChunkMetadatas belong to the seriesPath + */ public static List<ChunkMetaData> loadChunkMetadataFromTsFileResource( TsFileResource resource, Path seriesPath, QueryContext context) throws IOException { - List<ChunkMetaData> currentChunkMetaDataList; + List<ChunkMetaData> chunkMetaDataList; if (resource == null) { return new ArrayList<>(); } if (resource.isClosed()) { - currentChunkMetaDataList = DeviceMetaDataCache.getInstance().get(resource, seriesPath); + chunkMetaDataList = DeviceMetaDataCache.getInstance().get(resource, seriesPath); } else { - currentChunkMetaDataList = resource.getChunkMetaDataList(); + chunkMetaDataList = resource.getChunkMetaDataList(); } List<Modification> pathModifications = context.getPathModifications(resource.getModFile(), seriesPath.getFullPath()); if (!pathModifications.isEmpty()) { - QueryUtils.modifyChunkMetaData(currentChunkMetaDataList, pathModifications); + QueryUtils.modifyChunkMetaData(chunkMetaDataList, pathModifications); } - for (ChunkMetaData data : currentChunkMetaDataList) { + for (ChunkMetaData data : chunkMetaDataList) { TsFileSequenceReader tsFileSequenceReader = FileReaderManager.getInstance().get(resource, resource.isClosed()); data.setChunkLoader(new DiskChunkLoader(tsFileSequenceReader)); @@ -100,10 +120,16 @@ public class FileLoaderUtils { if (memChunks != null) { for (ReadOnlyMemChunk readOnlyMemChunk : memChunks) { if (!memChunks.isEmpty()) { - currentChunkMetaDataList.add(readOnlyMemChunk.getChunkMetaData()); + chunkMetaDataList.add(readOnlyMemChunk.getChunkMetaData()); } } } - return currentChunkMetaDataList; + + /* + * remove empty or invalid chunk metadata + */ + chunkMetaDataList.removeIf(chunkMetaData -> ( + chunkMetaData.getStartTime() > chunkMetaData.getEndTime())); + return chunkMetaDataList; } }
