You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ha...@apache.org on 2021/08/31 06:51:23 UTC
[iotdb] branch master updated: update design document of data query
(#3781)
This is an automated email from the ASF dual-hosted git repository.
haonan pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new e562445 update design document of data query (#3781)
e562445 is described below
commit e562445235a65c6c4c7100f970f1babc8779e905
Author: Hang Ji <55...@users.noreply.github.com>
AuthorDate: Tue Aug 31 14:50:55 2021 +0800
update design document of data query (#3781)
---
docs/SystemDesign/DataQuery/RawDataQuery.md | 26 ++++++++++------
docs/SystemDesign/DataQuery/SeriesReader.md | 36 +++++++++++++---------
docs/zh/SystemDesign/DataQuery/OrderByTimeQuery.md | 2 +-
docs/zh/SystemDesign/DataQuery/RawDataQuery.md | 26 ++++++++++------
docs/zh/SystemDesign/DataQuery/SeriesReader.md | 36 +++++++++++++---------
5 files changed, 79 insertions(+), 47 deletions(-)
diff --git a/docs/SystemDesign/DataQuery/RawDataQuery.md b/docs/SystemDesign/DataQuery/RawDataQuery.md
index 089d0b0..5ca11b9 100644
--- a/docs/SystemDesign/DataQuery/RawDataQuery.md
+++ b/docs/SystemDesign/DataQuery/RawDataQuery.md
@@ -120,7 +120,7 @@ Next, introduce the code of the consumer. The main logic of the consumer is to t
First introduce some important fields of consumer tasks
-* TreeSet\<Long\> timeHeap
+* TimeSelector timeHeap;
The smallest heap of timestamps for timestamp alignment
@@ -143,13 +143,13 @@ The `init ()` method was first called in the constructor of the consumer `RawQue
```
private void init() throws InterruptedException {
- timeHeap = new TreeSet<>();
+ timeHeap = new TimeSelector(seriesReaderList.size() << 1, ascending);
// Build producer tasks for each time series
for (int i = 0; i < seriesReaderList.size(); i++) {
ManagedSeriesReader reader = seriesReaderList.get(i);
reader.setHasRemaining(true);
reader.setManagedByQueryManager(true);
- pool.submit(new ReadTask(reader, blockingQueueArray[i]));
+ pool.submit(new ReadTask(reader, blockingQueueArray[i], paths.get(i).getFullPath()));
}
// Initialize the minimum heap and fill the cache for each time series
for (int i = 0; i < seriesReaderList.size(); i++) {
@@ -175,7 +175,15 @@ private void fillCache(int seriesIndex) throws InterruptedException {
// If it is a signal BatchData, set oMoreDataInQueue of the corresponding time series to false
if (batchData instanceof SignalBatchData) {
noMoreDataInQueueArray[seriesIndex] = true;
- }
+ } else if (batchData instanceof ExceptionBatchData) {
+ // exception happened in producer thread
+ ExceptionBatchData exceptionBatchData = (ExceptionBatchData) batchData;
+ LOGGER.error("exception happened in producer thread", exceptionBatchData.getException());
+ if (exceptionBatchData.getException() instanceof IOException) {
+ throw (IOException) exceptionBatchData.getException();
+ } else if (exceptionBatchData.getException() instanceof RuntimeException) {
+ throw (RuntimeException) exceptionBatchData.getException();
+ }
else {
// Cache the retrieved BatchData into cachedBatchDataArray
cachedBatchDataArray[seriesIndex] = batchData;
@@ -242,25 +250,25 @@ The specific query logic is not repeated here. You can refer to the query logic
## Include value filter + result set aligned by timestamp
-### org.apache.iotdb.db.query.dataset.EngineDataSetWithValueFilter
+### org.apache.iotdb.db.query.dataset.RawQueryDataSetWithValueFilter
`EngineDataSetWithValueFilter` implements query logic with value filter conditions.
Its query logic is to first generate a timestamp that meets the filtering conditions according to the query conditions, query the value of the projection column by the timestamp that meets the conditions, and then return the result set. It has four fields
-* private EngineTimeGenerator timeGenerator;
+* TimeGenerator timeGenerator;
Is used to generate timestamps that satisfy the filter
-* private List\<IReaderByTimestamp\> seriesReaderByTimestampList;
+* List\<IReaderByTimestamp\> seriesReaderByTimestampList;
Reader for each time series, used to get data based on timestamp
-* private boolean hasCachedRowRecord;
+* List<Boolean> cached;
Whether data rows are currently cached
-* private RowRecord cachedRowRecord;
+* List<RowRecord> cachedRowRecords
Data lines currently cached
diff --git a/docs/SystemDesign/DataQuery/SeriesReader.md b/docs/SystemDesign/DataQuery/SeriesReader.md
index 9ac7c1a..4cc83eb 100644
--- a/docs/SystemDesign/DataQuery/SeriesReader.md
+++ b/docs/SystemDesign/DataQuery/SeriesReader.md
@@ -131,10 +131,7 @@ while (aggregateReader.hasNextChunk()) {
```
// Get the value of the given timestamp, or null if none exists (requires that the //timestamp passed in is incremented)
-Object getValueInTimestamp(long timestamp) throws IOException;
-
-// Given a batch of timestamp values, return a batch of results (reduce the number //of method calls)
-Object[] getValuesInTimestamps(long[] timestamps) throws IOException;
+Object[] getValuesInTimestamps(long[] timestamps, int length) throws IOException;
```
#### General use process
@@ -142,11 +139,7 @@ Object[] getValuesInTimestamps(long[] timestamps) throws IOException;
This interface is used in queries with value filtering. After TimeGenerator generates a timestamp, use this interface to obtain the value corresponding to the timestamp.
```
-Object value = readerByTimestamp.getValueInTimestamp(timestamp);
-
-or
-
-Object[] values = readerByTimestamp.getValueInTimestamp(timestamps);
+Object[] values = readerByTimestamp.getValueInTimestamp(timestamps, length);
```
## Concrete implementation class
@@ -319,10 +312,18 @@ if (readPageData()) {
}
/*
- * If there is a chunk and a page, return page
+ * If there is a chunk, return chunk
+ */
+if (readChunkData()) {
+ hasCachedBatchData = true;
+ return true;
+}
+
+/*
+ * If there is a file and a chunk , return chunk
*/
-while (seriesReader.hasNextChunk()) {
- if (readPageData()) {
+while (seriesReader.hasNextFile()) {
+ if (readChunkData()) {
hasCachedBatchData = true;
return true;
}
@@ -349,7 +350,14 @@ if (readPageData(timestamp)) {
/*
* Determine if the next chunk has the current search time, skip it if it can
*/
-while (seriesReader.hasNextChunk()) {
+if (readChunkData(timestamp)) {
+ return true;
+}
+
+/*
+ * Determine if the next File has the current search time, skip it if it can
+ */
+while (seriesReader.hasNextFile()) {
Statistics statistics = seriesReader.currentChunkStatistics();
if (!satisfyTimeFilter(statistics)) {
seriesReader.skipCurrentChunk();
@@ -358,7 +366,7 @@ while (seriesReader.hasNextChunk()) {
/*
* The chunk cannot be skipped, continue to check the page in the chunk
*/
- if (readPageData(timestamp)) {
+ if (readChunkData(timestamp)) {
return true;
}
}
diff --git a/docs/zh/SystemDesign/DataQuery/OrderByTimeQuery.md b/docs/zh/SystemDesign/DataQuery/OrderByTimeQuery.md
index 9d5bd42..2c3a2eb 100644
--- a/docs/zh/SystemDesign/DataQuery/OrderByTimeQuery.md
+++ b/docs/zh/SystemDesign/DataQuery/OrderByTimeQuery.md
@@ -112,7 +112,7 @@
BatchData 类似一个二维数组的格式，封装了写入和读取的方法，对于倒序，抽象了下面的子类：
```java
-public class DescBatchData extends BatchData {
+public class DescReadBatchData extends BatchData {
//判断是否还有值，正序时候判断是否读到写入的 size 位置，倒序就是判断是否读到了 0 的位置
public boolean hasCurrent();
//指针移动到下一个数据，正序为 index+1 操作， 倒序为 index-1 操作
diff --git a/docs/zh/SystemDesign/DataQuery/RawDataQuery.md b/docs/zh/SystemDesign/DataQuery/RawDataQuery.md
index 7e5d21c..18a2ee5 100644
--- a/docs/zh/SystemDesign/DataQuery/RawDataQuery.md
+++ b/docs/zh/SystemDesign/DataQuery/RawDataQuery.md
@@ -123,7 +123,7 @@ public void run() {
先介绍消费者任务的一些重要字段
* ```
- TreeSet<Long> timeHeap
+ TimeSelector timeHeap;
```
时间戳的最小堆，用以实现时间戳对齐操作
@@ -152,13 +152,13 @@ public void run() {
```
private void init() throws InterruptedException {
- timeHeap = new TreeSet<>();
+ timeHeap = new TimeSelector(seriesReaderList.size() << 1, ascending);
// 为每个时间序列构建生产者任务
for (int i = 0; i < seriesReaderList.size(); i++) {
ManagedSeriesReader reader = seriesReaderList.get(i);
reader.setHasRemaining(true);
reader.setManagedByQueryManager(true);
- pool.submit(new ReadTask(reader, blockingQueueArray[i]));
+ pool.submit(new ReadTask(reader, blockingQueueArray[i], paths.get(i).getFullPath()));
}
// 初始化最小堆，填充每个时间序列对应的缓存
for (int i = 0; i < seriesReaderList.size(); i++) {
@@ -184,7 +184,15 @@ private void fillCache(int seriesIndex) throws InterruptedException {
// 如果是一个信号 BatchData，则将相应时间序列的 oMoreDataInQueue 置为 false
if (batchData instanceof SignalBatchData) {
noMoreDataInQueueArray[seriesIndex] = true;
- }
+ } else if (batchData instanceof ExceptionBatchData) {
+ // 当生产者线程发生异常时释放查询资源
+ ExceptionBatchData exceptionBatchData = (ExceptionBatchData) batchData;
+ LOGGER.error("exception happened in producer thread", exceptionBatchData.getException());
+ if (exceptionBatchData.getException() instanceof IOException) {
+ throw (IOException) exceptionBatchData.getException();
+ } else if (exceptionBatchData.getException() instanceof RuntimeException) {
+ throw (RuntimeException) exceptionBatchData.getException();
+ }
else {
// 将取出的 BatchData 放进 cachedBatchDataArray 缓存起来
cachedBatchDataArray[seriesIndex] = batchData;
@@ -251,32 +259,32 @@ for (int seriesIndex = 0; seriesIndex < seriesNum; seriesIndex++) {
## 包含值过滤条件 + 结果集按时间戳对齐
-### org.apache.iotdb.db.query.dataset.EngineDataSetWithValueFilter
+### org.apache.iotdb.db.query.dataset.RawQueryDataSetWithValueFilter
`EngineDataSetWithValueFilter`实现了有值过滤条件的查询逻辑。
它的查询逻辑是，首先根据查询条件生成满足过滤条件的时间戳，通过满足条件的时间戳查询投影列的值，然后返回结果集。它有四个字段
* ```
- private EngineTimeGenerator timeGenerator;
+ TimeGenerator timeGenerator;
```
是用来生成满足过滤条件的时间戳的
* ```
- private List<IReaderByTimestamp> seriesReaderByTimestampList;
+ List<IReaderByTimestamp> seriesReaderByTimestampList;
```
每个时间序列对应的 reader，用来根据时间戳获取数据
* ```
- private boolean hasCachedRowRecord;
+ List<Boolean> cached;
```
当前是否缓存了数据行
* ```
- private RowRecord cachedRowRecord;
+ List<RowRecord> cachedRowRecords
```
当前缓存的数据行
diff --git a/docs/zh/SystemDesign/DataQuery/SeriesReader.md b/docs/zh/SystemDesign/DataQuery/SeriesReader.md
index 244f42c..9946d02 100644
--- a/docs/zh/SystemDesign/DataQuery/SeriesReader.md
+++ b/docs/zh/SystemDesign/DataQuery/SeriesReader.md
@@ -130,11 +130,8 @@ while (aggregateReader.hasNextChunk()) {
#### 主要方法
```
-// 得到给定时间戳的值，如果不存在返回 null（要求传入的 timestamp 是递增的）
-Object getValueInTimestamp(long timestamp) throws IOException;
-
// 给定一批递增时间戳的值，返回一批结果（减少方法调用次数）
-Object[] getValuesInTimestamps(long[] timestamps) throws IOException;
+Object[] getValuesInTimestamps(long[] timestamps, int length) throws IOException;
```
#### 一般使用流程
@@ -142,11 +139,7 @@ Object[] getValuesInTimestamps(long[] timestamps) throws IOException;
该接口在带值过滤的查询中被使用，TimeGenerator 生成时间戳后，使用该接口获得该时间戳对应的 value
```
-Object value = readerByTimestamp.getValueInTimestamp(timestamp);
-
-or
-
-Object[] values = readerByTimestamp.getValueInTimestamp(timestamps);
+Object[] values = readerByTimestamp.getValueInTimestamp(timestamps, length);
```
## 具体实现类
@@ -319,10 +312,18 @@ if (readPageData()) {
}
/*
- * 如果有 chunk，并且有 page，返回 page
+ * 如果 SeriesReader 里还有 chunk，返回 chunk
+ */
+if (readChunkData()) {
+ hasCachedBatchData = true;
+ return true;
+}
+
+/*
+ * 如果有 File，并且有 chunk，返回 chunk
*/
-while (seriesReader.hasNextChunk()) {
- if (readPageData()) {
+while (seriesReader.hasNextFile()) {
+ if (readChunkData()) {
hasCachedBatchData = true;
return true;
}
@@ -349,7 +350,14 @@ if (readPageData(timestamp)) {
/*
* 判断下一个 chunk 有没有当前所查时间，能跳过就跳过
*/
-while (seriesReader.hasNextChunk()) {
+if (readChunkData(timestamp)) {
+ return true;
+}
+
+/*
+ * 判断下一个 File 有没有当前所查时间，能跳过就跳过
+ */
+while (seriesReader.hasNextFile()) {
Statistics statistics = seriesReader.currentChunkStatistics();
if (!satisfyTimeFilter(statistics)) {
seriesReader.skipCurrentChunk();
@@ -358,7 +366,7 @@ while (seriesReader.hasNextChunk()) {
/*
* chunk 不能跳过，继续到 chunk 里检查 page
*/
- if (readPageData(timestamp)) {
+ if (readChunkData(timestamp)) {
return true;
}
}