You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by qi...@apache.org on 2020/03/03 03:29:24 UTC
[incubator-iotdb] branch series_reader_doc created (now 30421a9)
This is an automated email from the ASF dual-hosted git repository.
qiaojialin pushed a change to branch series_reader_doc
in repository https://gitbox.apache.org/repos/asf/incubator-iotdb.git.
at 30421a9 update data query doc
This branch includes the following new commits:
new 30421a9 update data query doc
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
[incubator-iotdb] 01/01: update data query doc
Posted by qi...@apache.org.
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 <64...@qq.com>
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;
}
}