This is an automated email from the ASF dual-hosted git repository.
qiaojialin pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 6404d6d [IOTDB-298]Last time-value query (#821)
6404d6d is described below
commit 6404d6d27230234d28b73a8af1aa77a4fbbacdb1
Author: wshao08 <[email protected]>
AuthorDate: Tue Mar 3 08:56:12 2020 +0800
[IOTDB-298]Last time-value query (#821)
* Add Last query implement with cache
---
.../SystemDesign/5-DataQuery/1-DataQuery.md | 1 +
.../SystemDesign/5-DataQuery/6-LastQuery.md | 119 ++++++++++++++
.../2-DML (Data Manipulation Language).md | 33 ++++
.../5-Operation Manual/4-SQL Reference.md | 34 ++++
.../2-DML (Data Manipulation Language).md | 35 ++++
.../5-Operation Manual/4-SQL Reference.md | 33 ++++
.../org/apache/iotdb/db/qp/strategy/SqlBase.g4 | 10 ++
.../iotdb/db/engine/memtable/AbstractMemTable.java | 41 +----
.../engine/storagegroup/StorageGroupProcessor.java | 105 ++++++++++--
.../iotdb/db/metadata/mnode/InternalMNode.java | 1 +
.../apache/iotdb/db/metadata/mnode/LeafMNode.java | 29 ++++
.../org/apache/iotdb/db/metadata/mnode/MNode.java | 2 +-
.../apache/iotdb/db/qp/constant/SQLConstant.java | 2 +
.../apache/iotdb/db/qp/executor/PlanExecutor.java | 6 +-
.../org/apache/iotdb/db/qp/logical/Operator.java | 2 +-
.../iotdb/db/qp/logical/crud/SFWOperator.java | 8 +
.../iotdb/db/qp/logical/crud/SelectOperator.java | 7 +
.../iotdb/db/qp/physical/crud/BatchInsertPlan.java | 46 ++++++
.../iotdb/db/qp/physical/crud/InsertPlan.java | 13 ++
.../iotdb/db/qp/physical/crud/LastQueryPlan.java | 30 ++++
.../iotdb/db/qp/strategy/LogicalGenerator.java | 15 ++
.../iotdb/db/qp/strategy/PhysicalGenerator.java | 3 +
.../iotdb/db/query/executor/IQueryRouter.java | 7 +
.../iotdb/db/query/executor/LastQueryExecutor.java | 170 ++++++++++++++++++++
.../iotdb/db/query/executor/QueryRouter.java | 8 +
.../org/apache/iotdb/db/service/TSServiceImpl.java | 26 +--
.../org/apache/iotdb/db/utils/CommonUtils.java | 43 +++++
.../org/apache/iotdb/db/utils/FileLoaderUtils.java | 42 +++++
.../apache/iotdb/db/integration/IoTDBLastIT.java | 176 +++++++++++++++++++++
.../iotdb/db/metadata/MManagerAdvancedTest.java | 20 +++
.../apache/iotdb/db/qp/plan/PhysicalPlanTest.java | 26 +++
31 files changed, 1023 insertions(+), 70 deletions(-)
diff --git a/docs/Documentation-CHN/SystemDesign/5-DataQuery/1-DataQuery.md
b/docs/Documentation-CHN/SystemDesign/5-DataQuery/1-DataQuery.md
index 8c7ddb3..36e2aa8 100644
--- a/docs/Documentation-CHN/SystemDesign/5-DataQuery/1-DataQuery.md
+++ b/docs/Documentation-CHN/SystemDesign/5-DataQuery/1-DataQuery.md
@@ -37,3 +37,4 @@
* [原始数据查询](/#/SystemDesign/progress/chap5/sec3)
* [聚合查询](/#/SystemDesign/progress/chap5/sec4)
* [降采样查询](/#/SystemDesign/progress/chap5/sec5)
+* [最近时间戳查询](/#/SystemDesign/progress/chap5/sec6)
diff --git a/docs/Documentation-CHN/SystemDesign/5-DataQuery/6-LastQuery.md
b/docs/Documentation-CHN/SystemDesign/5-DataQuery/6-LastQuery.md
new file mode 100644
index 0000000..3394442
--- /dev/null
+++ b/docs/Documentation-CHN/SystemDesign/5-DataQuery/6-LastQuery.md
@@ -0,0 +1,119 @@
+<!--
+
+ 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.
+
+-->
+
+# 最近时间戳 Last 查询
+
+Last 查询的主要逻辑在 LastQueryExecutor
+
+* org.apache.iotdb.db.query.executor.LastQueryExecutor
+
+Last查询对每个指定的时间序列执行`calculateLastPairForOneSeries`方法。
+
+## 读取MNode缓存数据
+
+我们在需要查询的时间序列所对应的MNode结构中添加Last数据缓存。`calculateLastPairForOneSeries`方法对于某个时间序列的Last查询，首先尝试读取MNode中的缓存数据。
+```
+try {
+ node =
MManager.getInstance().getDeviceNodeWithAutoCreateStorageGroup(seriesPath.toString());
+} catch (MetadataException e) {
+ throw new QueryProcessException(e);
+}
+if (((LeafMNode) node).getCachedLast() != null) {
+ return ((LeafMNode) node).getCachedLast();
+}
+```
+如果发现缓存没有被写入过，则执行下面的标准查询流程读取TsFile数据。
+
+## Last标准查询流程
+
+Last标准查询流程需要遍历所有的顺序文件和乱序文件得到查询结果，最后将查询结果写回到MNode缓存。算法中对顺序文件和乱序文件分别进行处理。
+-
顺序文件由于是对其写入时间已经排好序，因此直接使用`loadChunkMetadataFromTsFileResource`方法取出最后一个`ChunkMetadata`，通过`ChunkMetadata`的统计数据得到最大时间戳和对应的值。
+ ```
+ if (!seqFileResources.isEmpty()) {
+ List<ChunkMetaData> chunkMetadata =
+ FileLoaderUtils.loadChunkMetadataFromTsFileResource(
+ seqFileResources.get(seqFileResources.size() - 1), seriesPath,
context);
+ if (!chunkMetadata.isEmpty()) {
+ ChunkMetaData lastChunkMetaData =
chunkMetadata.get(chunkMetadata.size() - 1);
+ Statistics chunkStatistics = lastChunkMetaData.getStatistics();
+ resultPair =
+ constructLastPair(
+ chunkStatistics.getEndTime(), chunkStatistics.getLastValue(),
tsDataType);
+ }
+ }
+ ```
+-
乱序文件则需要遍历所有的`ChunkMetadata`结构得到最大时间戳数据。需要注意的是当多个`ChunkMetadata`拥有相同的时间戳时，我们取`version`值最大的`ChunkMatadata`中的数据作为Last的结果。
+
+ ```
+ long version = 0;
+ for (TsFileResource resource : unseqFileResources) {
+ if (resource.getEndTimeMap().get(seriesPath.getDevice()) <
resultPair.getTimestamp()) {
+ break;
+ }
+ List<ChunkMetaData> chunkMetadata =
+ FileLoaderUtils.loadChunkMetadataFromTsFileResource(resource,
seriesPath, context);
+ for (ChunkMetaData chunkMetaData : chunkMetadata) {
+ if (chunkMetaData.getEndTime() == resultPair.getTimestamp()
+ && chunkMetaData.getVersion() > version) {
+ Statistics chunkStatistics = chunkMetaData.getStatistics();
+ resultPair =
+ constructLastPair(
+ chunkStatistics.getEndTime(),
chunkStatistics.getLastValue(), tsDataType);
+ version = chunkMetaData.getVersion();
+ }
+ }
+ }
+ ```
+ - 最后将查询结果写入到MNode的Last缓存
+ ```
+ ((LeafMNode) node).updateCachedLast(resultPair, false, Long.MIN_VALUE);
+ ```
+
+## Last 缓存更新策略
+
+Last缓存更新的逻辑位于`LeafMNode`的`updateCachedLast`方法内，这里引入两个额外的参数`highPriorityUpdate`和`latestFlushTime`。`highPriorityUpdate`用来表示本次更新是否是高优先级的，新数据写入而导致的缓存更新都被认为是高优先级更新，而查询时更新缓存默认为低优先级更新。`latestFlushTime`用来记录当前已被写回到磁盘的数据的最大时间戳。
+
+缓存更新的策略如下：
+
+1. 当缓存中没有记录时，对于查询到的Last数据，将查询的结果直接写入到缓存中。
+2. 当缓存中没有记录时，对于写入的最新数据如果时间戳大于或等于`latestFlushTime`，则将写入的数据写入到缓存中。
+3.
当缓存中已有记录时，根据查询或写入的数据时间戳与当前缓存中时间戳作对比。写入的数据具有高优先级，时间戳不小于缓存记录则更新缓存；查询出的数据低优先级，必须大于缓存记录的时间戳才更新缓存。
+
+具体代码如下
+```
+public synchronized void updateCachedLast(
+ TimeValuePair timeValuePair, boolean highPriorityUpdate, Long
latestFlushedTime) {
+ if (timeValuePair == null || timeValuePair.getValue() == null) return;
+
+ if (cachedLastValuePair == null) {
+ // If no cached last, (1) a last query (2) an unseq insertion or (3) a
seq insertion will update cache.
+ if (!highPriorityUpdate || latestFlushedTime <=
timeValuePair.getTimestamp()) {
+ cachedLastValuePair =
+ new TimeValuePair(timeValuePair.getTimestamp(),
timeValuePair.getValue());
+ }
+ } else if (timeValuePair.getTimestamp() >
cachedLastValuePair.getTimestamp()
+ || (timeValuePair.getTimestamp() == cachedLastValuePair.getTimestamp()
+ && highPriorityUpdate)) {
+ cachedLastValuePair.setTimestamp(timeValuePair.getTimestamp());
+ cachedLastValuePair.setValue(timeValuePair.getValue());
+ }
+}
+```
\ No newline at end of file
diff --git a/docs/Documentation-CHN/UserGuide/5-Operation Manual/2-DML (Data
Manipulation Language).md b/docs/Documentation-CHN/UserGuide/5-Operation
Manual/2-DML (Data Manipulation Language).md
index 872df0f..2b5647c 100644
--- a/docs/Documentation-CHN/UserGuide/5-Operation Manual/2-DML (Data
Manipulation Language).md
+++ b/docs/Documentation-CHN/UserGuide/5-Operation Manual/2-DML (Data
Manipulation Language).md
@@ -276,6 +276,39 @@ GROUP BY的SELECT子句里的查询路径必须是聚合函数，否则系统将
<center><img style="width:100%; max-width:800px; max-height:600px;
margin-left:auto; margin-right:auto; display:block;"
src="https://user-images.githubusercontent.com/16079446/69116099-0b715300-0ac6-11ea-8074-84e04797b8c7.png";></center>
+### 最近时间戳数据查询
+
+对应的SQL语句是：
+
+```
+select last <Path> [COMMA <Path>]* from < PrefixPath > [COMMA < PrefixPath >]*
<DISABLE ALIGN>
+```
+其含义是：
+
+查询时间序列prefixPath.path中最近时间戳的数据
+
+下面的例子中查询时间序列root.ln.wf01.wt01.status最近时间戳的数据:
+```
+select last status from root.ln.wf01.wt01 disable align
+```
+结果集为以下的形式返回：
+```
+| Time | Path | Value |
+| --- | ----------------------- | ----- |
+| 5 | root.ln.wf01.wt01.status| 100 |
+```
+
+假设root.ln.wf01.wt01中包含多列数据，如id, status, temperature，下面的例子将会把这几列数据在最近时间戳的记录同时返回：
+```
+select last id, status, temperature from root.ln.wf01 disable align
+
+| Time | Path | Value |
+| --- | ---------------------------- | ----- |
+| 5 | root.ln.wf01.wt01.id | 10 |
+| 7 | root.ln.wf01.wt01.status | true |
+| 9 | root.ln.wf01.wt01.temperature| 35.7 |
+```
+
## 数据维护
diff --git a/docs/Documentation-CHN/UserGuide/5-Operation Manual/4-SQL
Reference.md b/docs/Documentation-CHN/UserGuide/5-Operation Manual/4-SQL
Reference.md
index 4d3a530..240c35f 100644
--- a/docs/Documentation-CHN/UserGuide/5-Operation Manual/4-SQL Reference.md
+++ b/docs/Documentation-CHN/UserGuide/5-Operation Manual/4-SQL Reference.md
@@ -446,6 +446,40 @@ root.sg1.d0.s0 is INT32 while root.sg2.d3.s0 is FLOAT.
```
+* Last语句
+
+Last 语句返回所要查询时间序列的最近时间戳的一条数据
+
+```
+SELECT LAST <SelectClause> FROM <FromClause> <DisableAlignClause>
+Select Clause : <Path> [COMMA <Path>]*
+FromClause : < PrefixPath > [COMMA < PrefixPath >]*
+DisableAlignClause : [DISABLE ALIGN]
+
+Eg. SELECT LAST s1 FROM root.sg.d1 disable align
+Eg. SELECT LAST s1, s2 FROM root.sg.d1 disable align
+Eg. SELECT LAST s1 FROM root.sg.d1, root.sg.d2 disable align
+
+规则:
+1. 需要满足PrefixPath.Path 为一条完整的时间序列，即 <PrefixPath> + <Path> = <Timeseries>
+
+2. SELECT LAST 语句不支持过滤条件.
+
+3. 结果集以"disable align"的形式返回，表现为总是包含三列的表格。
+例如 "select last s1, s2 from root.sg.d1, root.sg.d2 disable align", 结果集返回如下：
+
+| Time | Path | Value |
+| --- | ------------ | ----- |
+| 5 | root.sg.d1.s1| 100 |
+| 2 | root.sg.d1.s2| 400 |
+| 4 | root.sg.d2.s1| 250 |
+| 9 | root.sg.d2.s2| 600 |
+
+4. SELECT LAST 查询语句要是总是和末尾的disable align在一起使用。如果用户不熟悉SELECT
LAST的语法或者忘记在末尾添加"disable align"，IoTDB 也会接受不包含"disable align"的SQL语句并且仍以"disable
align"的形式返回结果集。
+例如用户输入 "select last s1 from root.sg.d1" 所得到的查询结果与 "select last s1 from
root.sg.d1 disable align". 的结果是完全相同的。
+
+```
+
### 数据库管理语句
* 创建用户
diff --git a/docs/Documentation/UserGuide/5-Operation Manual/2-DML (Data
Manipulation Language).md b/docs/Documentation/UserGuide/5-Operation
Manual/2-DML (Data Manipulation Language).md
index 2986d4a..9184550 100644
--- a/docs/Documentation/UserGuide/5-Operation Manual/2-DML (Data Manipulation
Language).md
+++ b/docs/Documentation/UserGuide/5-Operation Manual/2-DML (Data Manipulation
Language).md
@@ -243,6 +243,41 @@ The path after SELECT in GROUP BY statement must be
aggregate function, otherwis
<center><img style="width:100%; max-width:800px; max-height:600px;
margin-left:auto; margin-right:auto; display:block;"
src="https://user-images.githubusercontent.com/16079446/69116099-0b715300-0ac6-11ea-8074-84e04797b8c7.png";></center>
+### Last timestamp Query
+In scenarios when IoT devices updates data in a fast manner, users are more
interested in the most recent record of IoT devices.
+The LAST query is to return the most recent value of the given timeseries in a
time-value pair format.
+
+The SQL statement is:
+
+```
+select last <Path> [COMMA <Path>]* from < PrefixPath > [COMMA < PrefixPath >]*
<DISABLE ALIGN>
+```
+which means:
+
+Query and return the data with the largest timestamp of timeseries
prefixPath.path.
+
+In the following example, we queries the latest record of timeseries
root.ln.wf01.wt01.status:
+```
+select last status from root.ln.wf01.wt01 disable align
+```
+The result will be returned in a three column table format.
+```
+| Time | Path | Value |
+| --- | ----------------------- | ----- |
+| 5 | root.ln.wf01.wt01.status| 100 |
+```
+If the path root.ln.wf01.wt01 has multiple columns, for example id, status and
temperature, the following case will return records of all the three
measurements with the largest timestamp.
+```
+select last id, status, temperature from root.ln.wf01.wt01 disable align
+
+| Time | Path | Value |
+| --- | ---------------------------- | ----- |
+| 5 | root.ln.wf01.wt01.id | 10 |
+| 7 | root.ln.wf01.wt01.status | true |
+| 9 | root.ln.wf01.wt01.temperature| 35.7 |
+```
+
+
### Automated Fill
In the actual use of IoTDB, when doing the query operation of timeseries,
situations where the value is null at some time points may appear, which will
obstruct the further analysis by users. In order to better reflect the degree
of data change, users expect missing values to be automatically filled.
Therefore, the IoTDB system introduces the function of Automated Fill.
diff --git a/docs/Documentation/UserGuide/5-Operation Manual/4-SQL Reference.md
b/docs/Documentation/UserGuide/5-Operation Manual/4-SQL Reference.md
index c4fa9b9..f7f5ab2 100644
--- a/docs/Documentation/UserGuide/5-Operation Manual/4-SQL Reference.md
+++ b/docs/Documentation/UserGuide/5-Operation Manual/4-SQL Reference.md
@@ -457,6 +457,39 @@ You could expect a table like:
```
+* Select Last Record Statement
+
+The LAST function returns the last time-value pair of the given timeseries.
Currently filters are not supported in LAST queries.
+
+```
+SELECT LAST <SelectClause> FROM <FromClause> <DisableAlignClause>
+Select Clause : <Path> [COMMA <Path>]*
+FromClause : < PrefixPath > [COMMA < PrefixPath >]*
+DisableAlignClause : [DISABLE ALIGN]
+
+Eg. SELECT LAST s1 FROM root.sg.d1 disable align
+Eg. SELECT LAST s1, s2 FROM root.sg.d1 disable align
+Eg. SELECT LAST s1 FROM root.sg.d1, root.sg.d2 disable align
+
+Rules:
+1. the statement needs to satisfy this constraint: <PrefixPath> + <Path> =
<Timeseries>
+
+2. The result set of last query will always be displayed in a
"disable-aligned" format showed below.
+For example, "select last s1, s2 from root.sg.d1, root.sg.d2 disable align",
the query result would be:
+
+| Time | Path | Value |
+| --- | ------------ | ----- |
+| 5 | root.sg.d1.s1| 100 |
+| 2 | root.sg.d1.s2| 400 |
+| 4 | root.sg.d2.s1| 250 |
+| 9 | root.sg.d2.s2| 600 |
+
+3. LAST query syntax is expecting users to write a "diable align" keyword at
the end of the query.
+However, as it is a unique SQL syntax in IoTDB, IoTDB accepts LAST queries
without "disable align" and treats them as "disable align" ones.
+Query like "select last s1 from root.sg.d1" will be parsed exactly the same as
"select last s1 from root.sg.d1 disable align".
+
+```
+
### Database Management Statement
* Create User
diff --git a/server/src/main/antlr4/org/apache/iotdb/db/qp/strategy/SqlBase.g4
b/server/src/main/antlr4/org/apache/iotdb/db/qp/strategy/SqlBase.g4
index 9e547e3..55943fc 100644
--- a/server/src/main/antlr4/org/apache/iotdb/db/qp/strategy/SqlBase.g4
+++ b/server/src/main/antlr4/org/apache/iotdb/db/qp/strategy/SqlBase.g4
@@ -93,6 +93,7 @@ selectElements
: functionCall (COMMA functionCall)* #functionElement
| suffixPath (COMMA suffixPath)* #selectElement
| STRING_LITERAL (COMMA STRING_LITERAL)* #selectConstElement
+ | lastClause #lastElement
;
functionCall
@@ -111,6 +112,10 @@ functionName
| LAST_VALUE
;
+lastClause
+ : LAST suffixPath (COMMA suffixPath)*
+ ;
+
attributeClauses
: DATATYPE OPERATOR_EQ dataType COMMA ENCODING OPERATOR_EQ encoding (COMMA
(COMPRESSOR | COMPRESSION) OPERATOR_EQ compressor=propertyValue)? (COMMA
property)*
;
@@ -149,6 +154,7 @@ specialClause
: specialLimit
| groupByClause specialLimit?
| fillClause slimitClause? alignByDeviceClauseOrDisableAlign?
+ | alignByDeviceClauseOrDisableAlign?
;
specialLimit
@@ -720,6 +726,10 @@ LAST_VALUE
: L A S T UNDERLINE V A L U E
;
+LAST
+ : L A S T
+ ;
+
DISABLE
: D I S A B L E
;
diff --git
a/server/src/main/java/org/apache/iotdb/db/engine/memtable/AbstractMemTable.java
b/server/src/main/java/org/apache/iotdb/db/engine/memtable/AbstractMemTable.java
index 35d7a66..3bd46dd 100644
---
a/server/src/main/java/org/apache/iotdb/db/engine/memtable/AbstractMemTable.java
+++
b/server/src/main/java/org/apache/iotdb/db/engine/memtable/AbstractMemTable.java
@@ -32,6 +32,7 @@ import org.apache.iotdb.db.qp.constant.SQLConstant;
import org.apache.iotdb.db.qp.physical.crud.BatchInsertPlan;
import org.apache.iotdb.db.qp.physical.crud.InsertPlan;
import org.apache.iotdb.db.rescon.TVListAllocator;
+import org.apache.iotdb.db.utils.CommonUtils;
import org.apache.iotdb.db.utils.MemUtils;
import org.apache.iotdb.db.utils.datastructure.TVList;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
@@ -89,7 +90,7 @@ public abstract class AbstractMemTable implements IMemTable {
try {
for (int i = 0; i < insertPlan.getValues().length; i++) {
- Object value = parseValue(insertPlan.getDataTypes()[i],
insertPlan.getValues()[i]);
+ Object value = CommonUtils.parseValue(insertPlan.getDataTypes()[i],
insertPlan.getValues()[i]);
write(insertPlan.getDeviceId(), insertPlan.getMeasurements()[i],
insertPlan.getDataTypes()[i], insertPlan.getTime(), value);
}
@@ -100,44 +101,6 @@ public abstract class AbstractMemTable implements
IMemTable {
}
}
- private static Object parseValue(TSDataType dataType, String value) throws
QueryProcessException {
- try {
- switch (dataType) {
- case BOOLEAN:
- value = value.toLowerCase();
- if (SQLConstant.BOOLEAN_FALSE_NUM.equals(value) ||
SQLConstant.BOOLEN_FALSE.equals(value)) {
- return false;
- }
- if (SQLConstant.BOOLEAN_TRUE_NUM.equals(value) ||
SQLConstant.BOOLEN_TRUE.equals(value)) {
- return true;
- }
- throw new QueryProcessException("The BOOLEAN should be true/TRUE,
false/FALSE or 0/1");
- case INT32:
- return Integer.parseInt(value);
- case INT64:
- return Long.parseLong(value);
- case FLOAT:
- return Float.parseFloat(value);
- case DOUBLE:
- return Double.parseDouble(value);
- case TEXT:
- if ((value.startsWith(SQLConstant.QUOTE) &&
value.endsWith(SQLConstant.QUOTE))
- || (value.startsWith(SQLConstant.DQUOTE) &&
value.endsWith(SQLConstant.DQUOTE))) {
- if (value.length() == 1) {
- return new Binary(value);
- } else {
- return new Binary(value.substring(1, value.length() - 1));
- }
- }
- throw new QueryProcessException("The TEXT data type should be
covered by \" or '");
- default:
- throw new QueryProcessException("Unsupported data type:" + dataType);
- }
- } catch (NumberFormatException e) {
- throw new QueryProcessException(e.getMessage());
- }
- }
-
@Override
public void insertBatch(BatchInsertPlan batchInsertPlan, int start, int end)
throws QueryProcessException {
diff --git
a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessor.java
b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessor.java
index edaa0c8..210afa6 100755
---
a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessor.java
+++
b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessor.java
@@ -69,6 +69,8 @@ import org.apache.iotdb.db.exception.query.OutOfTTLException;
import org.apache.iotdb.db.exception.query.QueryProcessException;
import
org.apache.iotdb.db.exception.storageGroup.StorageGroupProcessorException;
import org.apache.iotdb.db.metadata.MManager;
+import org.apache.iotdb.db.metadata.mnode.LeafMNode;
+import org.apache.iotdb.db.metadata.mnode.MNode;
import org.apache.iotdb.db.qp.physical.crud.BatchInsertPlan;
import org.apache.iotdb.db.qp.physical.crud.DeletePlan;
import org.apache.iotdb.db.qp.physical.crud.InsertPlan;
@@ -124,7 +126,7 @@ public class StorageGroupProcessor {
* a read write lock for guaranteeing concurrent safety when accessing all
fields in this class
* (i.e., schema, (un)sequenceFileList, work(un)SequenceTsFileProcessor,
* closing(Un)SequenceTsFileProcessor, latestTimeForEachDevice, and
- * latestFlushedTimeForEachDevice)
+ * partitionLatestFlushedTimeForEachDevice)
*/
private final ReadWriteLock insertLock = new ReentrantReadWriteLock();
/**
@@ -166,18 +168,24 @@ public class StorageGroupProcessor {
/*
* time partition id -> map, which contains
* device -> global latest timestamp of each device latestTimeForEachDevice
caches non-flushed
- * changes upon timestamps of each device, and is used to update
latestFlushedTimeForEachDevice
+ * changes upon timestamps of each device, and is used to update
partitionLatestFlushedTimeForEachDevice
* when a flush is issued.
*/
private Map<Long, Map<String, Long>> latestTimeForEachDevice = new
HashMap<>();
/**
* time partition id -> map, which contains device -> largest timestamp of
the latest memtable to
- * be submitted to asyncTryToFlush latestFlushedTimeForEachDevice determines
whether a data point
+ * be submitted to asyncTryToFlush partitionLatestFlushedTimeForEachDevice
determines whether a data point
* should be put into a sequential file or an unsequential file. Data of
some device with
* timestamp less than or equals to the device's latestFlushedTime should go
into an unsequential
* file.
*/
- private Map<Long, Map<String, Long>> latestFlushedTimeForEachDevice = new
HashMap<>();
+ private Map<Long, Map<String, Long>> partitionLatestFlushedTimeForEachDevice
= new HashMap<>();
+ /**
+ * global mapping of device -> largest timestamp of the latest memtable to *
be submitted to
+ * asyncTryToFlush, globalLatestFlushedTimeForEachDevice is utilized to
maintain global
+ * latestFlushedTime of devices and will be updated along with
partitionLatestFlushedTimeForEachDevice
+ */
+ private Map<String, Long> globalLatestFlushedTimeForEachDevice = new
HashMap<>();
private String storageGroupName;
private File storageGroupSysDir;
/**
@@ -304,8 +312,16 @@ public class StorageGroupProcessor {
if (timePartitionId != -1) {
latestTimeForEachDevice.computeIfAbsent(timePartitionId, l -> new
HashMap<>())
.putAll(resource.getEndTimeMap());
- latestFlushedTimeForEachDevice.computeIfAbsent(timePartitionId, id ->
new HashMap<>())
+
partitionLatestFlushedTimeForEachDevice.computeIfAbsent(timePartitionId, id ->
new HashMap<>())
.putAll(resource.getEndTimeMap());
+
+ for (Map.Entry<String, Long> mapEntry :
resource.getEndTimeMap().entrySet()) {
+ if
(!globalLatestFlushedTimeForEachDevice.containsKey(mapEntry.getKey())
+ || globalLatestFlushedTimeForEachDevice.get(mapEntry.getKey())
+ < mapEntry.getValue()) {
+ globalLatestFlushedTimeForEachDevice.put(mapEntry.getKey(),
mapEntry.getValue());
+ }
+ }
}
}
}
@@ -493,12 +509,12 @@ public class StorageGroupProcessor {
long timePartitionId = fromTimeToTimePartition(insertPlan.getTime());
latestTimeForEachDevice.computeIfAbsent(timePartitionId, l -> new
HashMap<>())
.putIfAbsent(insertPlan.getDeviceId(), Long.MIN_VALUE);
- latestFlushedTimeForEachDevice.computeIfAbsent(timePartitionId, id ->
new HashMap<>())
+ partitionLatestFlushedTimeForEachDevice.computeIfAbsent(timePartitionId,
id -> new HashMap<>())
.putIfAbsent(insertPlan.getDeviceId(), Long.MIN_VALUE);
// insert to sequence or unSequence file
insertToTsFileProcessor(insertPlan,
- insertPlan.getTime() >
latestFlushedTimeForEachDevice.get(timePartitionId)
+ insertPlan.getTime() >
partitionLatestFlushedTimeForEachDevice.get(timePartitionId)
.get(insertPlan.getDeviceId()));
} finally {
writeUnlock();
@@ -533,7 +549,7 @@ public class StorageGroupProcessor {
// before time partition
long beforeTimePartition =
fromTimeToTimePartition(batchInsertPlan.getTimes()[before]);
// init map
- long lastFlushTime = latestFlushedTimeForEachDevice.
+ long lastFlushTime = partitionLatestFlushedTimeForEachDevice.
computeIfAbsent(beforeTimePartition, id -> new HashMap<>()).
computeIfAbsent(batchInsertPlan.getDeviceId(), id -> Long.MIN_VALUE);
// if is sequence
@@ -550,7 +566,7 @@ public class StorageGroupProcessor {
// re initialize
before = loc;
beforeTimePartition = curTimePartition;
- lastFlushTime = latestFlushedTimeForEachDevice.
+ lastFlushTime = partitionLatestFlushedTimeForEachDevice.
computeIfAbsent(beforeTimePartition, id -> new HashMap<>()).
computeIfAbsent(batchInsertPlan.getDeviceId(), id ->
Long.MIN_VALUE);
isSequence = false;
@@ -624,6 +640,13 @@ public class StorageGroupProcessor {
latestTimeForEachDevice.get(timePartitionId)
.put(batchInsertPlan.getDeviceId(), batchInsertPlan.getTimes()[end -
1]);
}
+ long globalLatestFlushedTime =
+ globalLatestFlushedTimeForEachDevice.computeIfAbsent(
+ batchInsertPlan.getDeviceId(), k -> Long.MIN_VALUE);
+ tryToUpdateBatchInsertLastCache(batchInsertPlan, globalLatestFlushedTime);
+ if (globalLatestFlushedTime < batchInsertPlan.getMaxTime())
+ globalLatestFlushedTimeForEachDevice.put(
+ batchInsertPlan.getDeviceId(), batchInsertPlan.getMaxTime());
// check memtable size and may async try to flush the work memtable
if (tsFileProcessor.shouldFlush()) {
@@ -631,6 +654,22 @@ public class StorageGroupProcessor {
}
}
+ public void tryToUpdateBatchInsertLastCache(BatchInsertPlan plan, Long
latestFlushedTime)
+ throws QueryProcessException {
+ try {
+ MNode node =
+
MManager.getInstance().getDeviceNodeWithAutoCreateStorageGroup(plan.getDeviceId());
+ String[] measurementList = plan.getMeasurements();
+ for (int i = 0; i < measurementList.length; i++) {
+ // Update cached last value with high priority
+ MNode measurementNode = node.getChild(measurementList[i]);
+ ((LeafMNode) measurementNode)
+ .updateCachedLast(plan.composeLastTimeValuePair(i), true,
latestFlushedTime);
+ }
+ } catch (MetadataException e) {
+ throw new QueryProcessException(e);
+ }
+ }
private void insertToTsFileProcessor(InsertPlan insertPlan, boolean sequence)
throws QueryProcessException {
@@ -654,6 +693,13 @@ public class StorageGroupProcessor {
latestTimeForEachDevice.get(timePartitionId)
.put(insertPlan.getDeviceId(), insertPlan.getTime());
}
+ long globalLatestFlushTime =
+ globalLatestFlushedTimeForEachDevice.computeIfAbsent(
+ insertPlan.getDeviceId(), k -> Long.MIN_VALUE);
+ tryToUpdateInsertLastCache(insertPlan, globalLatestFlushTime);
+ if (result && globalLatestFlushTime < insertPlan.getTime()) {
+ globalLatestFlushedTimeForEachDevice.put(insertPlan.getDeviceId(),
insertPlan.getTime());
+ }
// check memtable size and may asyncTryToFlush the work memtable
if (tsFileProcessor.shouldFlush()) {
@@ -661,6 +707,23 @@ public class StorageGroupProcessor {
}
}
+ public void tryToUpdateInsertLastCache(InsertPlan plan, Long
latestFlushedTime)
+ throws QueryProcessException {
+ try {
+ MNode node =
+
MManager.getInstance().getDeviceNodeWithAutoCreateStorageGroup(plan.getDeviceId());
+ String[] measurementList = plan.getMeasurements();
+ for (int i = 0; i < measurementList.length; i++) {
+ // Update cached last value with high priority
+ MNode measurementNode = node.getChild(measurementList[i]);
+ ((LeafMNode) measurementNode)
+ .updateCachedLast(plan.composeTimeValuePair(i), true,
latestFlushedTime);
+ }
+ } catch (MetadataException e) {
+ throw new QueryProcessException(e);
+ }
+ }
+
private TsFileProcessor getOrCreateTsFileProcessor(long timeRangeId, boolean
sequence) {
TsFileProcessor tsFileProcessor = null;
try {
@@ -879,7 +942,8 @@ public class StorageGroupProcessor {
this.workUnsequenceTsFileProcessors.clear();
this.sequenceFileTreeSet.clear();
this.unSequenceFileList.clear();
- this.latestFlushedTimeForEachDevice.clear();
+ this.partitionLatestFlushedTimeForEachDevice.clear();
+ this.globalLatestFlushedTimeForEachDevice.clear();
this.latestTimeForEachDevice.clear();
} finally {
writeUnlock();
@@ -1265,9 +1329,13 @@ public class StorageGroupProcessor {
}
for (Entry<String, Long> entry : curPartitionDeviceLatestTime.entrySet()) {
- latestFlushedTimeForEachDevice
+ partitionLatestFlushedTimeForEachDevice
.computeIfAbsent(processor.getTimeRangeId(), id -> new HashMap<>())
.put(entry.getKey(), entry.getValue());
+ if (!globalLatestFlushedTimeForEachDevice.containsKey(entry.getKey())
+ || globalLatestFlushedTimeForEachDevice.get(entry.getKey()) <
entry.getValue()) {
+ globalLatestFlushedTimeForEachDevice.put(entry.getKey(),
entry.getValue());
+ }
}
return true;
}
@@ -1495,7 +1563,7 @@ public class StorageGroupProcessor {
* <p>
* Secondly, execute the loading process by the type.
* <p>
- * Finally, update the latestTimeForEachDevice and
latestFlushedTimeForEachDevice.
+ * Finally, update the latestTimeForEachDevice and
partitionLatestFlushedTimeForEachDevice.
* @param newTsFileResource tsfile resource
* @UsedBy sync module.
*/
@@ -1528,7 +1596,7 @@ public class StorageGroupProcessor {
* <p>
* Secondly, execute the loading process by the type.
* <p>
- * Finally, update the latestTimeForEachDevice and
latestFlushedTimeForEachDevice.
+ * Finally, update the latestTimeForEachDevice and
partitionLatestFlushedTimeForEachDevice.
*
* @param newTsFileResource tsfile resource
* @UsedBy load external tsfile module
@@ -1671,7 +1739,7 @@ public class StorageGroupProcessor {
}
/**
- * Update latest time in latestTimeForEachDevice and
latestFlushedTimeForEachDevice.
+ * Update latest time in latestTimeForEachDevice and
partitionLatestFlushedTimeForEachDevice.
*
* @UsedBy sync module, load external tsfile module.
*/
@@ -1686,15 +1754,19 @@ public class StorageGroupProcessor {
latestTimeForEachDevice.get(timePartitionId).put(device, endTime);
}
- Map<String, Long> latestFlushTimeForPartition =
latestFlushedTimeForEachDevice
+ Map<String, Long> latestFlushTimeForPartition =
partitionLatestFlushedTimeForEachDevice
.getOrDefault(timePartitionId, new HashMap<>());
if (!latestFlushTimeForPartition.containsKey(device)
|| latestFlushTimeForPartition.get(device) < endTime) {
- latestFlushedTimeForEachDevice
+ partitionLatestFlushedTimeForEachDevice
.computeIfAbsent(timePartitionId, id -> new HashMap<String,
Long>())
.put(device, endTime);
}
+ if (!globalLatestFlushedTimeForEachDevice.containsKey(device)
+ || globalLatestFlushedTimeForEachDevice.get(device) < endTime) {
+ globalLatestFlushedTimeForEachDevice.put(device, endTime);
+ }
}
}
@@ -1914,5 +1986,4 @@ public class StorageGroupProcessor {
boolean call(TsFileProcessor caller);
}
-
}
diff --git
a/server/src/main/java/org/apache/iotdb/db/metadata/mnode/InternalMNode.java
b/server/src/main/java/org/apache/iotdb/db/metadata/mnode/InternalMNode.java
index 18d56df..a215d45 100644
--- a/server/src/main/java/org/apache/iotdb/db/metadata/mnode/InternalMNode.java
+++ b/server/src/main/java/org/apache/iotdb/db/metadata/mnode/InternalMNode.java
@@ -20,6 +20,7 @@ package org.apache.iotdb.db.metadata.mnode;
import java.util.LinkedHashMap;
import java.util.Map;
+
import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
public class InternalMNode extends MNode {
diff --git
a/server/src/main/java/org/apache/iotdb/db/metadata/mnode/LeafMNode.java
b/server/src/main/java/org/apache/iotdb/db/metadata/mnode/LeafMNode.java
index e74c41c..0edf768 100644
--- a/server/src/main/java/org/apache/iotdb/db/metadata/mnode/LeafMNode.java
+++ b/server/src/main/java/org/apache/iotdb/db/metadata/mnode/LeafMNode.java
@@ -23,6 +23,7 @@ import java.util.Map;
import org.apache.iotdb.tsfile.file.metadata.enums.CompressionType;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.file.metadata.enums.TSEncoding;
+import org.apache.iotdb.tsfile.read.TimeValuePair;
import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
public class LeafMNode extends MNode {
@@ -34,6 +35,8 @@ public class LeafMNode extends MNode {
*/
private MeasurementSchema schema;
+ private TimeValuePair cachedLastValuePair = null;
+
public LeafMNode(MNode parent, String name, TSDataType dataType, TSEncoding
encoding,
CompressionType type, Map<String, String> props) {
super(parent, name);
@@ -74,4 +77,30 @@ public class LeafMNode extends MNode {
public MeasurementSchema getSchema() {
return schema;
}
+
+ public TimeValuePair getCachedLast() {
+ return cachedLastValuePair;
+ }
+
+ public synchronized void updateCachedLast(
+ TimeValuePair timeValuePair, boolean highPriorityUpdate, Long
latestFlushedTime) {
+ if (timeValuePair == null || timeValuePair.getValue() == null) return;
+
+ if (cachedLastValuePair == null) {
+ // If no cached last, (1) a last query (2) an unseq insertion or (3) a
seq insertion will update cache.
+ if (!highPriorityUpdate || latestFlushedTime <=
timeValuePair.getTimestamp()) {
+ cachedLastValuePair =
+ new TimeValuePair(timeValuePair.getTimestamp(),
timeValuePair.getValue());
+ }
+ } else if (timeValuePair.getTimestamp() >
cachedLastValuePair.getTimestamp()
+ || (timeValuePair.getTimestamp() == cachedLastValuePair.getTimestamp()
+ && highPriorityUpdate)) {
+ cachedLastValuePair.setTimestamp(timeValuePair.getTimestamp());
+ cachedLastValuePair.setValue(timeValuePair.getValue());
+ }
+ }
+
+ public void resetCache() {
+ cachedLastValuePair = null;
+ }
}
\ No newline at end of file
diff --git a/server/src/main/java/org/apache/iotdb/db/metadata/mnode/MNode.java
b/server/src/main/java/org/apache/iotdb/db/metadata/mnode/MNode.java
index 3ac7310..60008bf 100644
--- a/server/src/main/java/org/apache/iotdb/db/metadata/mnode/MNode.java
+++ b/server/src/main/java/org/apache/iotdb/db/metadata/mnode/MNode.java
@@ -113,4 +113,4 @@ public abstract class MNode implements Serializable {
public void setName(String name) {
this.name = name;
}
-}
\ No newline at end of file
+}
diff --git
a/server/src/main/java/org/apache/iotdb/db/qp/constant/SQLConstant.java
b/server/src/main/java/org/apache/iotdb/db/qp/constant/SQLConstant.java
index 108c850..19b5909 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/constant/SQLConstant.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/constant/SQLConstant.java
@@ -62,6 +62,8 @@ public class SQLConstant {
public static final String FIRST_VALUE = "first_value";
public static final String LAST_VALUE = "last_value";
+ public static final String LAST = "last";
+
public static final String COUNT = "count";
public static final String AVG = "avg";
public static final String SUM = "sum";
diff --git
a/server/src/main/java/org/apache/iotdb/db/qp/executor/PlanExecutor.java
b/server/src/main/java/org/apache/iotdb/db/qp/executor/PlanExecutor.java
index 01c72a7..6da6a00 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/executor/PlanExecutor.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/executor/PlanExecutor.java
@@ -80,6 +80,7 @@ import org.apache.iotdb.db.qp.physical.crud.GroupByPlan;
import org.apache.iotdb.db.qp.physical.crud.InsertPlan;
import org.apache.iotdb.db.qp.physical.crud.QueryPlan;
import org.apache.iotdb.db.qp.physical.crud.RawDataQueryPlan;
+import org.apache.iotdb.db.qp.physical.crud.LastQueryPlan;
import org.apache.iotdb.db.qp.physical.crud.UpdatePlan;
import org.apache.iotdb.db.qp.physical.sys.AuthorPlan;
import org.apache.iotdb.db.qp.physical.sys.CountPlan;
@@ -241,6 +242,8 @@ public class PlanExecutor implements IPlanExecutor {
} else if (queryPlan instanceof FillQueryPlan) {
FillQueryPlan fillQueryPlan = (FillQueryPlan) queryPlan;
queryDataSet = queryRouter.fill(fillQueryPlan, context);
+ } else if (queryPlan instanceof LastQueryPlan) {
+ queryDataSet = queryRouter.lastQuery((LastQueryPlan) queryPlan,
context);
} else {
queryDataSet = queryRouter.rawDataQuery((RawDataQueryPlan) queryPlan,
context);
}
@@ -801,8 +804,7 @@ public class PlanExecutor implements IPlanExecutor {
}
}
return storageEngine.insertBatch(batchInsertPlan);
-
- } catch (StorageEngineException | MetadataException e) {
+ } catch (PathException | StorageEngineException | MetadataException e) {
throw new QueryProcessException(e);
}
}
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/logical/Operator.java
b/server/src/main/java/org/apache/iotdb/db/qp/logical/Operator.java
index b9a44f2..00ecc6f 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/logical/Operator.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/logical/Operator.java
@@ -74,6 +74,6 @@ public abstract class Operator {
DELETE_ROLE, GRANT_ROLE_PRIVILEGE, REVOKE_ROLE_PRIVILEGE, LIST_USER,
LIST_ROLE,
LIST_USER_PRIVILEGE, LIST_ROLE_PRIVILEGE, LIST_USER_ROLES, LIST_ROLE_USERS,
GRANT_WATERMARK_EMBEDDING, REVOKE_WATERMARK_EMBEDDING,
- TTL, DELETE_STORAGE_GROUP, LOAD_CONFIGURATION, SHOW, LOAD_FILES,
REMOVE_FILE, MOVE_FILE
+ TTL, DELETE_STORAGE_GROUP, LOAD_CONFIGURATION, SHOW, LOAD_FILES,
REMOVE_FILE, MOVE_FILE, LAST
}
}
diff --git
a/server/src/main/java/org/apache/iotdb/db/qp/logical/crud/SFWOperator.java
b/server/src/main/java/org/apache/iotdb/db/qp/logical/crud/SFWOperator.java
index e675bfd..a6c0c24 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/logical/crud/SFWOperator.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/logical/crud/SFWOperator.java
@@ -32,6 +32,7 @@ public abstract class SFWOperator extends RootOperator {
private FromOperator fromOperator;
private FilterOperator filterOperator;
private boolean hasAggregation = false;
+ private boolean lastQuery = false;
public SFWOperator(int tokenIntType) {
super(tokenIntType);
@@ -58,6 +59,9 @@ public abstract class SFWOperator extends RootOperator {
if (!sel.getAggregations().isEmpty()) {
hasAggregation = true;
}
+ if (sel.isLastQuery()) {
+ lastQuery = true;
+ }
}
public FilterOperator getFilterOperator() {
@@ -84,4 +88,8 @@ public abstract class SFWOperator extends RootOperator {
public boolean hasAggregation() {
return hasAggregation;
}
+
+ public boolean isLastQuery() {
+ return lastQuery;
+ }
}
diff --git
a/server/src/main/java/org/apache/iotdb/db/qp/logical/crud/SelectOperator.java
b/server/src/main/java/org/apache/iotdb/db/qp/logical/crud/SelectOperator.java
index 902dc38..a0e566e 100644
---
a/server/src/main/java/org/apache/iotdb/db/qp/logical/crud/SelectOperator.java
+++
b/server/src/main/java/org/apache/iotdb/db/qp/logical/crud/SelectOperator.java
@@ -30,6 +30,7 @@ public final class SelectOperator extends Operator {
private List<Path> suffixList;
private List<String> aggregations;
+ private boolean lastQuery;
/**
* init with tokenIntType, default operatorType is
<code>OperatorType.SELECT</code>.
@@ -39,6 +40,7 @@ public final class SelectOperator extends Operator {
operatorType = OperatorType.SELECT;
suffixList = new ArrayList<>();
aggregations = new ArrayList<>();
+ lastQuery = false;
}
public void addSelectPath(Path suffixPath) {
@@ -50,6 +52,10 @@ public final class SelectOperator extends Operator {
aggregations.add(aggregation);
}
+ public void setLastQuery() {
+ lastQuery = true;
+ }
+
public List<String> getAggregations() {
return this.aggregations;
}
@@ -66,4 +72,5 @@ public final class SelectOperator extends Operator {
return suffixList;
}
+ public boolean isLastQuery() {return this.lastQuery; }
}
diff --git
a/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/BatchInsertPlan.java
b/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/BatchInsertPlan.java
index fe1ee39..11382a4 100644
---
a/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/BatchInsertPlan.java
+++
b/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/BatchInsertPlan.java
@@ -22,14 +22,23 @@ import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
import java.util.Set;
+
import org.apache.iotdb.db.qp.logical.Operator.OperatorType;
import org.apache.iotdb.db.qp.physical.PhysicalPlan;
import org.apache.iotdb.db.utils.QueryDataSetUtils;
import org.apache.iotdb.tsfile.exception.write.UnSupportedDataTypeException;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.read.TimeValuePair;
import org.apache.iotdb.tsfile.read.common.Path;
import org.apache.iotdb.tsfile.utils.Binary;
import org.apache.iotdb.tsfile.utils.BytesUtils;
+import org.apache.iotdb.tsfile.utils.TsPrimitiveType;
+import org.apache.iotdb.tsfile.utils.TsPrimitiveType.TsBinary;
+import org.apache.iotdb.tsfile.utils.TsPrimitiveType.TsBoolean;
+import org.apache.iotdb.tsfile.utils.TsPrimitiveType.TsDouble;
+import org.apache.iotdb.tsfile.utils.TsPrimitiveType.TsFloat;
+import org.apache.iotdb.tsfile.utils.TsPrimitiveType.TsInt;
+import org.apache.iotdb.tsfile.utils.TsPrimitiveType.TsLong;
public class BatchInsertPlan extends PhysicalPlan {
@@ -288,6 +297,43 @@ public class BatchInsertPlan extends PhysicalPlan {
return maxTime;
}
+ public TimeValuePair composeLastTimeValuePair(int measurementIndex) {
+ if (measurementIndex >= columns.length) {
+ return null;
+ }
+ TsPrimitiveType value;
+ switch (dataTypes[measurementIndex]) {
+ case INT32:
+ int[] intValues = (int[]) columns[measurementIndex];
+ value = new TsInt(intValues[end - 1]);
+ break;
+ case INT64:
+ long[] longValues = (long[]) columns[measurementIndex];
+ value = new TsLong(longValues[end - 1]);
+ break;
+ case FLOAT:
+ float[] floatValues = (float[]) columns[measurementIndex];
+ value = new TsFloat(floatValues[end - 1]);
+ break;
+ case DOUBLE:
+ double[] doubleValues = (double[]) columns[measurementIndex];
+ value = new TsDouble(doubleValues[end - 1]);
+ break;
+ case BOOLEAN:
+ boolean[] boolValues = (boolean[]) columns[measurementIndex];
+ value = new TsBoolean(boolValues[end - 1]);
+ break;
+ case TEXT:
+ Binary[] binaryValues = (Binary[]) columns[measurementIndex];
+ value = new TsBinary(binaryValues[end - 1]);
+ break;
+ default:
+ throw new UnSupportedDataTypeException(
+ String.format("Data type %s is not supported.",
dataTypes[measurementIndex]));
+ }
+ return new TimeValuePair(times[end - 1], value);
+ }
+
public long[] getTimes() {
return times;
}
diff --git
a/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/InsertPlan.java
b/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/InsertPlan.java
index 048370b..6544298 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/InsertPlan.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/InsertPlan.java
@@ -23,12 +23,17 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Objects;
+
+import org.apache.iotdb.db.exception.query.QueryProcessException;
import org.apache.iotdb.db.qp.logical.Operator;
import org.apache.iotdb.db.qp.logical.Operator.OperatorType;
import org.apache.iotdb.db.qp.physical.PhysicalPlan;
+import org.apache.iotdb.db.utils.CommonUtils;
import org.apache.iotdb.db.utils.TestOnly;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.read.TimeValuePair;
import org.apache.iotdb.tsfile.read.common.Path;
+import org.apache.iotdb.tsfile.utils.TsPrimitiveType;
import org.apache.iotdb.tsfile.write.record.TSRecord;
public class InsertPlan extends PhysicalPlan {
@@ -185,4 +190,12 @@ public class InsertPlan extends PhysicalPlan {
public String toString() {
return "deviceId: " + deviceId + ", time: " + time;
}
+
+ public TimeValuePair composeTimeValuePair(int measurementIndex) throws
QueryProcessException {
+ if (measurementIndex >= values.length) {
+ return null;
+ }
+ Object value = CommonUtils.parseValue(dataTypes[measurementIndex],
values[measurementIndex]);
+ return new TimeValuePair(time,
TsPrimitiveType.getByType(dataTypes[measurementIndex], value));
+ }
}
diff --git
a/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/LastQueryPlan.java
b/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/LastQueryPlan.java
new file mode 100644
index 0000000..5ab7dc5
--- /dev/null
+++
b/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/LastQueryPlan.java
@@ -0,0 +1,30 @@
+/*
+ * 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.qp.physical.crud;
+
+import org.apache.iotdb.db.qp.logical.Operator;
+
+public class LastQueryPlan extends RawDataQueryPlan {
+
+ public LastQueryPlan() {
+ super();
+ setOperatorType(Operator.OperatorType.LAST);
+ }
+}
diff --git
a/server/src/main/java/org/apache/iotdb/db/qp/strategy/LogicalGenerator.java
b/server/src/main/java/org/apache/iotdb/db/qp/strategy/LogicalGenerator.java
index b67f048..4d5257d 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/strategy/LogicalGenerator.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/strategy/LogicalGenerator.java
@@ -93,6 +93,7 @@ import
org.apache.iotdb.db.qp.strategy.SqlBaseParser.InsertColumnSpecContext;
import org.apache.iotdb.db.qp.strategy.SqlBaseParser.InsertStatementContext;
import org.apache.iotdb.db.qp.strategy.SqlBaseParser.InsertValuesSpecContext;
import org.apache.iotdb.db.qp.strategy.SqlBaseParser.LimitClauseContext;
+import org.apache.iotdb.db.qp.strategy.SqlBaseParser.LastClauseContext;
import org.apache.iotdb.db.qp.strategy.SqlBaseParser.ListAllRoleOfUserContext;
import org.apache.iotdb.db.qp.strategy.SqlBaseParser.ListAllUserOfRoleContext;
import org.apache.iotdb.db.qp.strategy.SqlBaseParser.ListPrivilegesRoleContext;
@@ -988,6 +989,20 @@ public class LogicalGenerator extends SqlBaseBaseListener {
}
@Override
+ public void enterLastElement(SqlBaseParser.LastElementContext ctx) {
+ super.enterLastElement(ctx);
+ selectOp = new SelectOperator(SQLConstant.TOK_SELECT);
+ selectOp.setLastQuery();
+ LastClauseContext lastClauseContext = ctx.lastClause();
+ List<SuffixPathContext> suffixPaths = lastClauseContext.suffixPath();
+ for (SuffixPathContext suffixPath : suffixPaths) {
+ Path path = parseSuffixPath(suffixPath);
+ selectOp.addSelectPath(path);
+ }
+ queryOp.setSelectOperator(selectOp);
+ }
+
+ @Override
public void enterSetCol(SetColContext ctx) {
super.enterSetCol(ctx);
selectOp.addSelectPath(parseSuffixPath(ctx.suffixPath()));
diff --git
a/server/src/main/java/org/apache/iotdb/db/qp/strategy/PhysicalGenerator.java
b/server/src/main/java/org/apache/iotdb/db/qp/strategy/PhysicalGenerator.java
index 49a21c2..c19d330 100644
---
a/server/src/main/java/org/apache/iotdb/db/qp/strategy/PhysicalGenerator.java
+++
b/server/src/main/java/org/apache/iotdb/db/qp/strategy/PhysicalGenerator.java
@@ -65,6 +65,7 @@ import org.apache.iotdb.db.qp.physical.crud.GroupByPlan;
import org.apache.iotdb.db.qp.physical.crud.InsertPlan;
import org.apache.iotdb.db.qp.physical.crud.QueryPlan;
import org.apache.iotdb.db.qp.physical.crud.RawDataQueryPlan;
+import org.apache.iotdb.db.qp.physical.crud.LastQueryPlan;
import org.apache.iotdb.db.qp.physical.sys.AuthorPlan;
import org.apache.iotdb.db.qp.physical.sys.CountPlan;
import org.apache.iotdb.db.qp.physical.sys.CreateTimeSeriesPlan;
@@ -236,6 +237,8 @@ public class PhysicalGenerator {
queryPlan = new AggregationPlan();
((AggregationPlan) queryPlan)
.setAggregations(queryOperator.getSelectOperator().getAggregations());
+ } else if (queryOperator.isLastQuery()) {
+ queryPlan = new LastQueryPlan();
} else {
queryPlan = new RawDataQueryPlan();
}
diff --git
a/server/src/main/java/org/apache/iotdb/db/query/executor/IQueryRouter.java
b/server/src/main/java/org/apache/iotdb/db/query/executor/IQueryRouter.java
index 31f7081..5b4215d 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/executor/IQueryRouter.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/executor/IQueryRouter.java
@@ -26,6 +26,7 @@ import org.apache.iotdb.db.qp.physical.crud.AggregationPlan;
import org.apache.iotdb.db.qp.physical.crud.FillQueryPlan;
import org.apache.iotdb.db.qp.physical.crud.GroupByPlan;
import org.apache.iotdb.db.qp.physical.crud.RawDataQueryPlan;
+import org.apache.iotdb.db.qp.physical.crud.LastQueryPlan;
import org.apache.iotdb.db.query.context.QueryContext;
import
org.apache.iotdb.tsfile.exception.filter.QueryFilterOptimizationException;
import org.apache.iotdb.tsfile.read.query.dataset.QueryDataSet;
@@ -55,4 +56,10 @@ public interface IQueryRouter {
*/
QueryDataSet fill(FillQueryPlan fillQueryPlan, QueryContext context)
throws StorageEngineException, QueryProcessException, IOException;
+
+ /**
+ * Execute last query
+ */
+ QueryDataSet lastQuery(LastQueryPlan lastQueryPlan, QueryContext context)
+ throws StorageEngineException, QueryProcessException, IOException;
}
diff --git
a/server/src/main/java/org/apache/iotdb/db/query/executor/LastQueryExecutor.java
b/server/src/main/java/org/apache/iotdb/db/query/executor/LastQueryExecutor.java
new file mode 100644
index 0000000..30940f5
--- /dev/null
+++
b/server/src/main/java/org/apache/iotdb/db/query/executor/LastQueryExecutor.java
@@ -0,0 +1,170 @@
+/*
+ * 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.executor;
+
+import static org.apache.iotdb.db.conf.IoTDBConstant.COLUMN_VALUE;
+import static org.apache.iotdb.db.conf.IoTDBConstant.COLUMN_TIMESERIES;
+
+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.exception.StorageEngineException;
+import org.apache.iotdb.db.exception.metadata.MetadataException;
+import org.apache.iotdb.db.exception.query.QueryProcessException;
+import org.apache.iotdb.db.metadata.MManager;
+import org.apache.iotdb.db.metadata.mnode.LeafMNode;
+import org.apache.iotdb.db.metadata.mnode.MNode;
+import org.apache.iotdb.db.qp.physical.crud.LastQueryPlan;
+import org.apache.iotdb.db.query.context.QueryContext;
+import org.apache.iotdb.db.query.control.FileReaderManager;
+import org.apache.iotdb.db.query.control.QueryResourceManager;
+import org.apache.iotdb.db.query.dataset.ListDataSet;
+import org.apache.iotdb.db.query.reader.chunk.DiskChunkLoader;
+import org.apache.iotdb.db.utils.FileLoaderUtils;
+import org.apache.iotdb.db.utils.QueryUtils;
+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.Field;
+import org.apache.iotdb.tsfile.read.common.Path;
+import org.apache.iotdb.tsfile.read.common.RowRecord;
+import org.apache.iotdb.tsfile.read.query.dataset.QueryDataSet;
+import org.apache.iotdb.tsfile.utils.Binary;
+import org.apache.iotdb.tsfile.utils.TsPrimitiveType;
+
+import java.io.IOException;
+import java.util.*;
+
+public class LastQueryExecutor {
+ private List<Path> selectedSeries;
+ private List<TSDataType> dataTypes;
+
+ public LastQueryExecutor(LastQueryPlan lastQueryPlan) {
+ this.selectedSeries = lastQueryPlan.getPaths();
+ this.dataTypes = lastQueryPlan.getDataTypes();
+ }
+
+ /**
+ * execute last function
+ *
+ * @param context query context
+ */
+ public QueryDataSet execute(QueryContext context)
+ throws StorageEngineException, IOException, QueryProcessException {
+
+ ListDataSet dataSet =
+ new ListDataSet(
+ Arrays.asList(new Path(COLUMN_TIMESERIES), new Path(COLUMN_VALUE)),
+ Arrays.asList(TSDataType.TEXT, TSDataType.TEXT));
+
+ for (int i = 0; i < selectedSeries.size(); i++) {
+ TimeValuePair lastTimeValuePair =
+ calculateLastPairForOneSeries(selectedSeries.get(i),
dataTypes.get(i), context);
+ if (lastTimeValuePair.getValue() != null) {
+ RowRecord resultRecord = new
RowRecord(lastTimeValuePair.getTimestamp());
+ Field pathField = new Field(TSDataType.TEXT);
+ pathField.setBinaryV(new Binary(selectedSeries.get(i).getFullPath()));
+ resultRecord.addField(pathField);
+
+ Field valueField = new Field(TSDataType.TEXT);
+ valueField.setBinaryV(new
Binary(lastTimeValuePair.getValue().getStringValue()));
+ resultRecord.addField(valueField);
+
+ dataSet.putRecord(resultRecord);
+ }
+ }
+
+ return dataSet;
+ }
+
+ /**
+ * get last result for one series
+ *
+ * @param context query context
+ * @return TimeValuePair
+ */
+ private TimeValuePair calculateLastPairForOneSeries(
+ Path seriesPath, TSDataType tsDataType, QueryContext context)
+ throws IOException, QueryProcessException, StorageEngineException {
+
+ // Retrieve last value from MNode
+ MNode node = null;
+ try {
+ node =
MManager.getInstance().getDeviceNodeWithAutoCreateStorageGroup(seriesPath.toString());
+ } catch (MetadataException e) {
+ throw new QueryProcessException(e);
+ }
+ if (((LeafMNode) node).getCachedLast() != null) {
+ return ((LeafMNode) node).getCachedLast();
+ }
+
+ QueryDataSource dataSource =
+ QueryResourceManager.getInstance().getQueryDataSource(seriesPath,
context, null);
+
+ List<TsFileResource> seqFileResources = dataSource.getSeqResources();
+ List<TsFileResource> unseqFileResources = dataSource.getUnseqResources();
+
+ TimeValuePair resultPair = new TimeValuePair(Long.MIN_VALUE, null);
+
+ if (!seqFileResources.isEmpty()) {
+ List<ChunkMetaData> chunkMetadata =
+ FileLoaderUtils.loadChunkMetadataFromTsFileResource(
+ seqFileResources.get(seqFileResources.size() - 1), seriesPath,
context);
+ if (!chunkMetadata.isEmpty()) {
+ ChunkMetaData lastChunkMetaData =
chunkMetadata.get(chunkMetadata.size() - 1);
+ Statistics chunkStatistics = lastChunkMetaData.getStatistics();
+ resultPair =
+ constructLastPair(
+ chunkStatistics.getEndTime(), chunkStatistics.getLastValue(),
tsDataType);
+ }
+ }
+
+ long version = 0;
+ for (TsFileResource resource : unseqFileResources) {
+ if (resource.getEndTimeMap().get(seriesPath.getDevice()) <
resultPair.getTimestamp()) {
+ break;
+ }
+ List<ChunkMetaData> chunkMetadata =
+ FileLoaderUtils.loadChunkMetadataFromTsFileResource(resource,
seriesPath, context);
+ for (ChunkMetaData chunkMetaData : chunkMetadata) {
+ if (chunkMetaData.getEndTime() == resultPair.getTimestamp()
+ && chunkMetaData.getVersion() > version) {
+ Statistics chunkStatistics = chunkMetaData.getStatistics();
+ resultPair =
+ constructLastPair(
+ chunkStatistics.getEndTime(),
chunkStatistics.getLastValue(), tsDataType);
+ version = chunkMetaData.getVersion();
+ }
+ }
+ }
+
+ // Update cached last value with low priority
+ ((LeafMNode) node).updateCachedLast(resultPair, false, Long.MIN_VALUE);
+ return resultPair;
+ }
+
+ private TimeValuePair constructLastPair(long timestamp, Object value,
TSDataType dataType) {
+ return new TimeValuePair(timestamp, TsPrimitiveType.getByType(dataType,
value));
+ }
+}
diff --git
a/server/src/main/java/org/apache/iotdb/db/query/executor/QueryRouter.java
b/server/src/main/java/org/apache/iotdb/db/query/executor/QueryRouter.java
index 3f27bb4..7d9c47d 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/executor/QueryRouter.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/executor/QueryRouter.java
@@ -28,6 +28,7 @@ import org.apache.iotdb.db.qp.physical.crud.AggregationPlan;
import org.apache.iotdb.db.qp.physical.crud.FillQueryPlan;
import org.apache.iotdb.db.qp.physical.crud.GroupByPlan;
import org.apache.iotdb.db.qp.physical.crud.RawDataQueryPlan;
+import org.apache.iotdb.db.qp.physical.crud.LastQueryPlan;
import org.apache.iotdb.db.query.context.QueryContext;
import org.apache.iotdb.db.query.dataset.groupby.GroupByWithValueFilterDataSet;
import
org.apache.iotdb.db.query.dataset.groupby.GroupByWithoutValueFilterDataSet;
@@ -149,4 +150,11 @@ public class QueryRouter implements IQueryRouter {
return fillQueryExecutor.execute(context);
}
+ @Override
+ public QueryDataSet lastQuery(LastQueryPlan lastQueryPlan, QueryContext
context)
+ throws StorageEngineException, QueryProcessException, IOException {
+ LastQueryExecutor lastQueryExecutor = new LastQueryExecutor(lastQueryPlan);
+ return lastQueryExecutor.execute(context);
+ }
+
}
diff --git
a/server/src/main/java/org/apache/iotdb/db/service/TSServiceImpl.java
b/server/src/main/java/org/apache/iotdb/db/service/TSServiceImpl.java
index 5b3da96..3386c40 100644
--- a/server/src/main/java/org/apache/iotdb/db/service/TSServiceImpl.java
+++ b/server/src/main/java/org/apache/iotdb/db/service/TSServiceImpl.java
@@ -23,6 +23,8 @@ import static
org.apache.iotdb.db.conf.IoTDBConstant.COLUMN_ROLE;
import static org.apache.iotdb.db.conf.IoTDBConstant.COLUMN_STORAGE_GROUP;
import static org.apache.iotdb.db.conf.IoTDBConstant.COLUMN_TTL;
import static org.apache.iotdb.db.conf.IoTDBConstant.COLUMN_USER;
+import static org.apache.iotdb.db.conf.IoTDBConstant.COLUMN_VALUE;
+import static org.apache.iotdb.db.conf.IoTDBConstant.COLUMN_TIMESERIES;
import java.io.IOException;
import java.nio.ByteBuffer;
@@ -65,11 +67,7 @@ import org.apache.iotdb.db.qp.executor.IPlanExecutor;
import org.apache.iotdb.db.qp.executor.PlanExecutor;
import org.apache.iotdb.db.qp.logical.Operator.OperatorType;
import org.apache.iotdb.db.qp.physical.PhysicalPlan;
-import org.apache.iotdb.db.qp.physical.crud.AlignByDevicePlan;
-import org.apache.iotdb.db.qp.physical.crud.BatchInsertPlan;
-import org.apache.iotdb.db.qp.physical.crud.DeletePlan;
-import org.apache.iotdb.db.qp.physical.crud.InsertPlan;
-import org.apache.iotdb.db.qp.physical.crud.QueryPlan;
+import org.apache.iotdb.db.qp.physical.crud.*;
import org.apache.iotdb.db.qp.physical.sys.AuthorPlan;
import org.apache.iotdb.db.qp.physical.sys.CreateTimeSeriesPlan;
import org.apache.iotdb.db.qp.physical.sys.DeleteStorageGroupPlan;
@@ -785,14 +783,13 @@ public class TSServiceImpl implements TSIService.Iface,
ServerContext {
QueryPlan plan = (QueryPlan) physicalPlan;
if (plan instanceof AlignByDevicePlan) {
getAlignByDeviceQueryHeaders((AlignByDevicePlan) plan, respColumns,
columnsTypes);
- // set dataTypeList in TSExecuteStatementResp. Note this is without
deduplication.
- resp.setColumns(respColumns);
- resp.setDataTypeList(columnsTypes);
+ } else if (plan instanceof LastQueryPlan) {
+ getLastQueryHeaders(plan, respColumns, columnsTypes);
} else {
getWideQueryHeaders(plan, respColumns, columnsTypes);
- resp.setColumns(respColumns);
- resp.setDataTypeList(columnsTypes);
}
+ resp.setColumns(respColumns);
+ resp.setDataTypeList(columnsTypes);
return resp;
}
@@ -929,6 +926,15 @@ public class TSServiceImpl implements TSIService.Iface,
ServerContext {
plan.setDataTypeConsistencyChecker(null);
}
+ private void getLastQueryHeaders(
+ QueryPlan plan, List<String> respColumns, List<String> columnTypes)
+ throws TException, QueryProcessException {
+ respColumns.add(COLUMN_TIMESERIES);
+ respColumns.add(COLUMN_VALUE);
+ columnTypes.add(TSDataType.TEXT.toString());
+ columnTypes.add(TSDataType.TEXT.toString());
+ }
+
@Override
public TSFetchResultsResp fetchResults(TSFetchResultsReq req) {
try {
diff --git a/server/src/main/java/org/apache/iotdb/db/utils/CommonUtils.java
b/server/src/main/java/org/apache/iotdb/db/utils/CommonUtils.java
index 8373e79..3f9e43a 100644
--- a/server/src/main/java/org/apache/iotdb/db/utils/CommonUtils.java
+++ b/server/src/main/java/org/apache/iotdb/db/utils/CommonUtils.java
@@ -22,7 +22,12 @@ import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
+
+import org.apache.iotdb.db.exception.query.QueryProcessException;
+import org.apache.iotdb.db.qp.constant.SQLConstant;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.fileSystem.FSFactoryProducer;
+import org.apache.iotdb.tsfile.utils.Binary;
public class CommonUtils {
@@ -61,4 +66,42 @@ public class CommonUtils {
return Files.walk(folder).filter(p -> p.toFile().isFile())
.mapToLong(p -> p.toFile().length()).sum();
}
+
+ public static Object parseValue(TSDataType dataType, String value) throws
QueryProcessException {
+ try {
+ switch (dataType) {
+ case BOOLEAN:
+ value = value.toLowerCase();
+ if (SQLConstant.BOOLEAN_FALSE_NUM.equals(value) ||
SQLConstant.BOOLEN_FALSE.equals(value)) {
+ return false;
+ }
+ if (SQLConstant.BOOLEAN_TRUE_NUM.equals(value) ||
SQLConstant.BOOLEN_TRUE.equals(value)) {
+ return true;
+ }
+ throw new QueryProcessException("The BOOLEAN should be true/TRUE,
false/FALSE or 0/1");
+ case INT32:
+ return Integer.parseInt(value);
+ case INT64:
+ return Long.parseLong(value);
+ case FLOAT:
+ return Float.parseFloat(value);
+ case DOUBLE:
+ return Double.parseDouble(value);
+ case TEXT:
+ if ((value.startsWith(SQLConstant.QUOTE) &&
value.endsWith(SQLConstant.QUOTE))
+ || (value.startsWith(SQLConstant.DQUOTE) &&
value.endsWith(SQLConstant.DQUOTE))) {
+ if (value.length() == 1) {
+ return new Binary(value);
+ } else {
+ return new Binary(value.substring(1, value.length() - 1));
+ }
+ }
+ throw new QueryProcessException("The TEXT data type should be
covered by \" or '");
+ default:
+ throw new QueryProcessException("Unsupported data type:" + dataType);
+ }
+ } catch (NumberFormatException e) {
+ throw new QueryProcessException(e.getMessage());
+ }
+ }
}
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 cd011ac..671e5d4 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
@@ -19,14 +19,22 @@
package org.apache.iotdb.db.utils;
import java.io.IOException;
+import java.util.ArrayList;
import java.util.List;
+import org.apache.iotdb.db.engine.cache.DeviceMetaDataCache;
+import org.apache.iotdb.db.engine.modification.Modification;
+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.reader.chunk.DiskChunkLoader;
import org.apache.iotdb.tsfile.file.metadata.ChunkGroupMetaData;
import org.apache.iotdb.tsfile.file.metadata.ChunkMetaData;
import org.apache.iotdb.tsfile.file.metadata.TsDeviceMetadata;
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;
public class FileLoaderUtils {
@@ -64,4 +72,38 @@ public class FileLoaderUtils {
}
}
}
+
+ public static List<ChunkMetaData> loadChunkMetadataFromTsFileResource(
+ TsFileResource resource, Path seriesPath, QueryContext context) 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 data : currentChunkMetaDataList) {
+ TsFileSequenceReader tsFileSequenceReader =
+ FileReaderManager.getInstance().get(resource, resource.isClosed());
+ data.setChunkLoader(new DiskChunkLoader(tsFileSequenceReader));
+ }
+ List<ReadOnlyMemChunk> memChunks = resource.getReadOnlyMemChunk();
+ if (memChunks != null) {
+ for (ReadOnlyMemChunk readOnlyMemChunk : memChunks) {
+ if (!memChunks.isEmpty()) {
+ currentChunkMetaDataList.add(readOnlyMemChunk.getChunkMetaData());
+ }
+ }
+ }
+ return currentChunkMetaDataList;
+ }
}
diff --git
a/server/src/test/java/org/apache/iotdb/db/integration/IoTDBLastIT.java
b/server/src/test/java/org/apache/iotdb/db/integration/IoTDBLastIT.java
new file mode 100644
index 0000000..c2cb34b
--- /dev/null
+++ b/server/src/test/java/org/apache/iotdb/db/integration/IoTDBLastIT.java
@@ -0,0 +1,176 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iotdb.db.integration;
+
+import static org.junit.Assert.fail;
+
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Statement;
+import org.apache.iotdb.db.exception.metadata.MetadataException;
+import org.apache.iotdb.db.exception.query.QueryProcessException;
+import org.apache.iotdb.db.metadata.MManager;
+import org.apache.iotdb.db.metadata.mnode.LeafMNode;
+import org.apache.iotdb.db.metadata.mnode.MNode;
+import org.apache.iotdb.db.utils.EnvironmentUtils;
+import org.apache.iotdb.db.utils.TestOnly;
+import org.apache.iotdb.jdbc.Config;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+public class IoTDBLastIT {
+
+ private static String[] dataSet = new String[]{
+ "SET STORAGE GROUP TO root.ln.wf01.wt01",
+ "CREATE TIMESERIES root.ln.wf01.wt01.status WITH DATATYPE=BOOLEAN,
ENCODING=PLAIN",
+ "CREATE TIMESERIES root.ln.wf01.wt01.temperature WITH DATATYPE=DOUBLE,
ENCODING=PLAIN",
+ "CREATE TIMESERIES root.ln.wf01.wt01.id WITH DATATYPE=INT32,
ENCODING=PLAIN",
+ "INSERT INTO root.ln.wf01.wt01(timestamp,temperature,status, id) "
+ + "values(100, 25.1, false, 7)",
+ "INSERT INTO root.ln.wf01.wt01(timestamp,temperature,status, id) "
+ + "values(200, 25.2, true, 8)",
+ "INSERT INTO root.ln.wf01.wt01(timestamp,temperature,status, id) "
+ + "values(300, 15.7, false, 9)",
+ "INSERT INTO root.ln.wf01.wt01(timestamp,temperature,status, id) "
+ + "values(400, 16.2, false, 6)",
+ "INSERT INTO root.ln.wf01.wt01(timestamp,temperature,status, id) "
+ + "values(500, 22.1, false, 5)",
+ "flush",
+ };
+
+ private static final String TIMESTAMP_STR = "Time";
+ private static final String TIMESEIRES_STR = "timeseries";
+ private static final String VALUE_STR = "value";
+
+ @Before
+ public void setUp() throws Exception {
+ EnvironmentUtils.closeStatMonitor();
+ EnvironmentUtils.envSetUp();
+ Class.forName(Config.JDBC_DRIVER_NAME);
+ prepareData();
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ EnvironmentUtils.cleanEnv();
+ }
+
+ @Test
+ public void lastCacheTest() throws SQLException {
+ String[] retArray1 =
+ new String[] {
+ "500,root.ln.wf01.wt01.temperature,22.1",
+ "500,root.ln.wf01.wt01.status,false",
+ "500,root.ln.wf01.wt01.id,5"
+ };
+ String[] retArray2 =
+ new String[] {
+ "700,root.ln.wf01.wt01.temperature,33.1",
+ "700,root.ln.wf01.wt01.status,false",
+ "700,root.ln.wf01.wt01.id,3"
+ };
+
+ try (Connection connection =
+ DriverManager.getConnection("jdbc:iotdb://127.0.0.1:6667/",
"root", "root");
+ Statement statement = connection.createStatement()) {
+
+ boolean hasResultSet =
+ statement.execute("select last temperature,status,id from
root.ln.wf01.wt01");
+
+ Assert.assertTrue(hasResultSet);
+ int cnt = 0;
+ try (ResultSet resultSet = statement.getResultSet()) {
+ while (resultSet.next()) {
+ String ans = resultSet.getString(TIMESTAMP_STR) + ","
+ + resultSet.getString(TIMESEIRES_STR) + ","
+ + resultSet.getString(VALUE_STR);
+ Assert.assertEquals(retArray1[cnt], ans);
+ cnt++;
+ }
+ }
+
+ MNode node = MManager.getInstance()
+
.getDeviceNodeWithAutoCreateStorageGroup("root.ln.wf01.wt01.temperature");
+ ((LeafMNode) node).resetCache();
+
+ hasResultSet = statement.execute(
+ "insert into root.ln.wf01.wt01(time, temperature, status, id)
values(700, 33.1, false, 3)");
+
+ // Last cache is updated with above insert sql
+ long time = ((LeafMNode) node).getCachedLast().getTimestamp();
+ Assert.assertEquals(time, 700);
+
+ hasResultSet = statement.execute("select last temperature,status,id from
root.ln.wf01.wt01");
+ Assert.assertTrue(hasResultSet);
+ cnt = 0;
+ try (ResultSet resultSet = statement.getResultSet()) {
+ while (resultSet.next()) {
+ String ans = resultSet.getString(TIMESTAMP_STR) + ","
+ + resultSet.getString(TIMESEIRES_STR) + ","
+ + resultSet.getString(VALUE_STR);
+ Assert.assertEquals(retArray2[cnt], ans);
+ cnt++;
+ }
+ }
+
+ hasResultSet = statement.execute(
+ "insert into root.ln.wf01.wt01(time, temperature, status, id)
values(600, 19.1, false, 1)");
+
+ // Last cache is not updated with above insert sql
+ time = ((LeafMNode) node).getCachedLast().getTimestamp();
+ Assert.assertEquals(time, 700);
+
+ hasResultSet = statement.execute("select last temperature,status,id from
root.ln.wf01.wt01");
+ Assert.assertTrue(hasResultSet);
+ cnt = 0;
+ try (ResultSet resultSet = statement.getResultSet()) {
+ while (resultSet.next()) {
+ String ans = resultSet.getString(TIMESTAMP_STR) + ","
+ + resultSet.getString(TIMESEIRES_STR) + ","
+ + resultSet.getString(VALUE_STR);
+ Assert.assertEquals(retArray2[cnt], ans);
+ cnt++;
+ }
+ }
+ } catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ private void prepareData() {
+ try (Connection connection = DriverManager
+ .getConnection(Config.IOTDB_URL_PREFIX + "127.0.0.1:6667/", "root",
+ "root");
+ Statement statement = connection.createStatement()) {
+
+
+ for (String sql : dataSet) {
+ statement.execute(sql);
+ }
+
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+ }
+}
diff --git
a/server/src/test/java/org/apache/iotdb/db/metadata/MManagerAdvancedTest.java
b/server/src/test/java/org/apache/iotdb/db/metadata/MManagerAdvancedTest.java
index 3b90be8..45beeb1 100644
---
a/server/src/test/java/org/apache/iotdb/db/metadata/MManagerAdvancedTest.java
+++
b/server/src/test/java/org/apache/iotdb/db/metadata/MManagerAdvancedTest.java
@@ -25,9 +25,12 @@ import java.io.IOException;
import java.util.List;
import org.apache.iotdb.db.exception.metadata.MetadataException;
import org.apache.iotdb.db.exception.storageGroup.StorageGroupException;
+import org.apache.iotdb.db.metadata.mnode.LeafMNode;
import org.apache.iotdb.db.metadata.mnode.MNode;
import org.apache.iotdb.db.utils.EnvironmentUtils;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.read.TimeValuePair;
+import org.apache.iotdb.tsfile.utils.TsPrimitiveType;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
@@ -117,4 +120,21 @@ public class MManagerAdvancedTest {
// ignore
}
}
+
+ @Test
+ public void testCachedLastTimeValue()
+ throws MetadataException, IOException, StorageGroupException {
+ mmanager.createTimeseries("root.vehicle.d2.s0", "DOUBLE", "RLE");
+
+ TimeValuePair tv1 = new TimeValuePair(1000,
TsPrimitiveType.getByType(TSDataType.DOUBLE, 1.0));
+ TimeValuePair tv2 = new TimeValuePair(2000,
TsPrimitiveType.getByType(TSDataType.DOUBLE, 3.0));
+ TimeValuePair tv3 = new TimeValuePair(1500,
TsPrimitiveType.getByType(TSDataType.DOUBLE, 2.5));
+ MNode node = mmanager.getNodeByPath("root.vehicle.d2.s0");
+ ((LeafMNode)node).updateCachedLast(tv1, true, Long.MIN_VALUE);
+ ((LeafMNode)node).updateCachedLast(tv2, true, Long.MIN_VALUE);
+ Assert.assertEquals(tv2.getTimestamp(),
((LeafMNode)node).getCachedLast().getTimestamp());
+ ((LeafMNode)node).updateCachedLast(tv3, true, Long.MIN_VALUE);
+ Assert.assertEquals(tv2.getTimestamp(),
((LeafMNode)node).getCachedLast().getTimestamp());
+ }
+
}
diff --git
a/server/src/test/java/org/apache/iotdb/db/qp/plan/PhysicalPlanTest.java
b/server/src/test/java/org/apache/iotdb/db/qp/plan/PhysicalPlanTest.java
index c2371c9..4a885b6 100644
--- a/server/src/test/java/org/apache/iotdb/db/qp/plan/PhysicalPlanTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/qp/plan/PhysicalPlanTest.java
@@ -38,6 +38,7 @@ import org.apache.iotdb.db.qp.physical.crud.FillQueryPlan;
import org.apache.iotdb.db.qp.physical.crud.GroupByPlan;
import org.apache.iotdb.db.qp.physical.crud.QueryPlan;
import org.apache.iotdb.db.qp.physical.crud.RawDataQueryPlan;
+import org.apache.iotdb.db.qp.physical.crud.LastQueryPlan;
import org.apache.iotdb.db.qp.physical.sys.AuthorPlan;
import org.apache.iotdb.db.qp.physical.sys.CreateTimeSeriesPlan;
import org.apache.iotdb.db.qp.physical.sys.DataAuthPlan;
@@ -607,4 +608,29 @@ public class PhysicalPlanTest {
Assert.assertEquals(1, plan.getDeduplicatedDataTypes().size());
Assert.assertEquals(new Path("root.vehicle.d1.s1"),
plan.getDeduplicatedPaths().get(0));
}
+
+ @Test
+ public void testLastPlanPaths() throws QueryProcessException {
+ String sqlStr1 = "SELECT last s1 FROM root.vehicle.d1";
+ String sqlStr2 = "SELECT last s1 FROM root.vehicle.d1, root.vehicle.d2";
+ PhysicalPlan plan1 = processor.parseSQLToPhysicalPlan(sqlStr1);
+ PhysicalPlan plan2 = processor.parseSQLToPhysicalPlan(sqlStr2);
+ Path path1 = new Path("root.vehicle.d1.s1");
+ Path path2 = new Path("root.vehicle.d2.s1");
+ assertEquals(1, plan1.getPaths().size());
+ assertEquals(path1.toString(), plan1.getPaths().get(0).toString());
+ assertEquals(2, plan2.getPaths().size());
+ assertEquals(path1.toString(), plan2.getPaths().get(0).toString());
+ assertEquals(path2.toString(), plan2.getPaths().get(1).toString());
+ }
+
+ @Test
+ public void testLastPlanDataTypes() throws QueryProcessException {
+ String sqlStr = "SELECT last s1 FROM root.vehicle.d1";
+ PhysicalPlan plan = processor.parseSQLToPhysicalPlan(sqlStr);
+
+ assertEquals(1, ((LastQueryPlan) plan).getDataTypes().size());
+ TSDataType dataType = ((LastQueryPlan) plan).getDataTypes().get(0);
+ assertEquals(TSDataType.FLOAT, dataType);
+ }
}
![https://user-images.githubusercontent.com/16079446/69116099-0b715300-0ac6-11ea-8074-84e04797b8c7.png"](https://user-images.githubusercontent.com/16079446/69116099-0b715300-0ac6-11ea-8074-84e04797b8c7.png"){kind=link}