You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@iotdb.apache.org by GitBox <gi...@apache.org> on 2020/02/17 18:35:37 UTC
[GitHub] [incubator-iotdb] wshao08 opened a new pull request #821:
[IOTDB-298]Last time-value query
wshao08 opened a new pull request #821: [IOTDB-298]Last time-value query
URL: https://github.com/apache/incubator-iotdb/pull/821
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] wshao08 commented on a change in pull request
#821: [IOTDB-298]Last time-value query
Posted by GitBox <gi...@apache.org>.
wshao08 commented on a change in pull request #821: [IOTDB-298]Last time-value query
URL: https://github.com/apache/incubator-iotdb/pull/821#discussion_r380597536
##########
File path: server/src/main/java/org/apache/iotdb/db/qp/logical/Operator.java
##########
@@ -68,7 +68,7 @@ public String toString() {
public enum OperatorType {
SFW, JOIN, UNION, FILTER, GROUPBY, ORDERBY, LIMIT, SELECT, SEQTABLESCAN, HASHTABLESCAN,
MERGEJOIN, FILEREAD, NULL, TABLESCAN, UPDATE, INSERT, BATCHINSERT, DELETE, BASIC_FUNC, IN, QUERY, MERGEQUERY,
- AGGREGATION, AUTHOR, FROM, FUNC, LOADDATA, METADATA, PROPERTY, INDEX, INDEXQUERY, FILL,
+ AGGREGATION, AUTHOR, FROM, FUNC, LOADDATA, METADATA, PROPERTY, INDEX, INDEXQUERY, FILL, LAST,
Review comment:
Modified
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] JackieTien97 commented on a change in pull
request #821: [IOTDB-298]Last time-value query
Posted by GitBox <gi...@apache.org>.
JackieTien97 commented on a change in pull request #821: [IOTDB-298]Last time-value query
URL: https://github.com/apache/incubator-iotdb/pull/821#discussion_r381024249
##########
File path: server/src/main/java/org/apache/iotdb/db/qp/executor/PlanExecutor.java
##########
@@ -680,6 +683,10 @@ public void insert(InsertPlan insertPlan) throws QueryProcessException {
}
insertPlan.setDataTypes(dataTypes);
storageEngine.insert(insertPlan);
+ for (int i = 0; i < measurementList.length; i++) {
+ MNode measurementNode = node.getChild(measurementList[i]);
+ measurementNode.updateCachedLast(insertPlan.composeTimeValuePair(i));
+ }
Review comment:
Why not combine this loop with the before one.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] wshao08 commented on a change in pull request
#821: [IOTDB-298]Last time-value query
Posted by GitBox <gi...@apache.org>.
wshao08 commented on a change in pull request #821: [IOTDB-298]Last time-value query
URL: https://github.com/apache/incubator-iotdb/pull/821#discussion_r380561185
##########
File path: server/src/main/java/org/apache/iotdb/db/metadata/MNode.java
##########
@@ -22,10 +22,15 @@
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.Map;
+
+import com.sun.rowset.internal.Row;
import org.apache.iotdb.db.conf.IoTDBConstant;
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.read.common.RowRecord;
Review comment:
Fixed
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] wshao08 commented on a change in pull request
#821: [IOTDB-298]Last time-value query
Posted by GitBox <gi...@apache.org>.
wshao08 commented on a change in pull request #821: [IOTDB-298]Last time-value query
URL: https://github.com/apache/incubator-iotdb/pull/821#discussion_r380561357
##########
File path: server/src/main/java/org/apache/iotdb/db/query/executor/LastQueryExecutor.java
##########
@@ -0,0 +1,201 @@
+/*
+ * 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 org.apache.iotdb.db.exception.StorageEngineException;
+import org.apache.iotdb.db.exception.path.PathException;
+import org.apache.iotdb.db.exception.query.QueryProcessException;
+import org.apache.iotdb.db.metadata.MManager;
+import org.apache.iotdb.db.metadata.MNode;
+import org.apache.iotdb.db.qp.constant.SQLConstant;
+import org.apache.iotdb.db.qp.physical.crud.AggregationPlan;
+import org.apache.iotdb.db.qp.physical.crud.LastQueryPlan;
+import org.apache.iotdb.db.query.aggregation.AggregateResult;
+import org.apache.iotdb.db.query.context.QueryContext;
+import org.apache.iotdb.db.query.control.QueryResourceManager;
+import org.apache.iotdb.db.query.dataset.SingleDataSet;
+import org.apache.iotdb.db.query.factory.AggreResultFactory;
+import org.apache.iotdb.db.query.reader.series.IAggregateReader;
+import org.apache.iotdb.db.query.reader.series.SeriesAggregateReader;
+import org.apache.iotdb.tsfile.exception.cache.CacheException;
+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.common.BatchData;
+import org.apache.iotdb.tsfile.read.common.Path;
+import org.apache.iotdb.tsfile.read.common.RowRecord;
+import org.apache.iotdb.tsfile.read.filter.basic.Filter;
+import org.apache.iotdb.tsfile.read.query.dataset.QueryDataSet;
+import org.apache.iotdb.tsfile.utils.TsPrimitiveType;
+import org.omg.CORBA.TIMEOUT;
Review comment:
Fixed, thanks
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] JackieTien97 commented on a change in pull
request #821: [IOTDB-298]Last time-value query
Posted by GitBox <gi...@apache.org>.
JackieTien97 commented on a change in pull request #821: [IOTDB-298]Last time-value query
URL: https://github.com/apache/incubator-iotdb/pull/821#discussion_r380531833
##########
File path: server/src/main/java/org/apache/iotdb/db/query/executor/LastQueryExecutor.java
##########
@@ -0,0 +1,201 @@
+/*
+ * 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 org.apache.iotdb.db.exception.StorageEngineException;
+import org.apache.iotdb.db.exception.path.PathException;
+import org.apache.iotdb.db.exception.query.QueryProcessException;
+import org.apache.iotdb.db.metadata.MManager;
+import org.apache.iotdb.db.metadata.MNode;
+import org.apache.iotdb.db.qp.constant.SQLConstant;
+import org.apache.iotdb.db.qp.physical.crud.AggregationPlan;
+import org.apache.iotdb.db.qp.physical.crud.LastQueryPlan;
+import org.apache.iotdb.db.query.aggregation.AggregateResult;
+import org.apache.iotdb.db.query.context.QueryContext;
+import org.apache.iotdb.db.query.control.QueryResourceManager;
+import org.apache.iotdb.db.query.dataset.SingleDataSet;
+import org.apache.iotdb.db.query.factory.AggreResultFactory;
+import org.apache.iotdb.db.query.reader.series.IAggregateReader;
+import org.apache.iotdb.db.query.reader.series.SeriesAggregateReader;
+import org.apache.iotdb.tsfile.exception.cache.CacheException;
+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.common.BatchData;
+import org.apache.iotdb.tsfile.read.common.Path;
+import org.apache.iotdb.tsfile.read.common.RowRecord;
+import org.apache.iotdb.tsfile.read.filter.basic.Filter;
+import org.apache.iotdb.tsfile.read.query.dataset.QueryDataSet;
+import org.apache.iotdb.tsfile.utils.TsPrimitiveType;
+import org.omg.CORBA.TIMEOUT;
+
+import java.io.IOException;
+import java.sql.Time;
+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 {
+
+ List<LastQueryResult> lastQueryResultList = new ArrayList<>();
+ for (int i = 0; i < selectedSeries.size(); i++) {
+ LastQueryResult lastQueryResult = calculateLastPairForOneSeries(selectedSeries.get(i), dataTypes.get(i), context);
+ lastQueryResultList.add(lastQueryResult);
+ }
+
+ RowRecord resultRecord = constructLastRowRecord(lastQueryResultList);
+ SingleDataSet dataSet = new SingleDataSet(selectedSeries, dataTypes);
+ dataSet.setRecord(resultRecord);
+ return dataSet;
+ }
+
+ /**
+ * get aggregation result for one series
+ *
+ * @param context query context
+ * @return AggregateResult list
+ */
+ private LastQueryResult calculateLastPairForOneSeries(
+ Path seriesPath, TSDataType tsDataType,
+ QueryContext context)
+ throws IOException, QueryProcessException, StorageEngineException {
+ LastQueryResult queryResult = new LastQueryResult();
+ MNode node = null;
+ try {
+ node = MManager.getInstance().getNodeByPathFromCache(seriesPath.toString());
+ } catch (PathException e) {
+ throw new QueryProcessException(e);
+ } catch (CacheException e) {
+ throw new QueryProcessException(e.getMessage());
+ }
+ if (node.getCachedLast() != null) {
+ queryResult.setPairResult(node.getCachedLast());
+ return queryResult;
+ }
+
+ // construct series reader without value filter
+ Filter timeFilter = null;
+ IAggregateReader seriesReader = new SeriesAggregateReader(
+ seriesPath, tsDataType, context, QueryResourceManager.getInstance()
+ .getQueryDataSource(seriesPath, context, timeFilter), timeFilter, null);
+
+ long maxTime = Long.MIN_VALUE;
+ while (seriesReader.hasNextChunk()) {
+ while (seriesReader.hasNextPage()) {
+ // cal by page data
+ while (seriesReader.hasNextOverlappedPage()) {
+ BatchData nextOverlappedPageData = seriesReader.nextOverlappedPage();
+ int maxIndex = nextOverlappedPageData.length() - 1;
+ if (maxIndex < 0) {
+ continue;
+ }
+ long time = nextOverlappedPageData.getTimeByIndex(maxIndex);
+ if (time > maxTime) {
+ maxTime = time;
+ queryResult.setPairResult(maxTime, nextOverlappedPageData.getValueInTimestamp(time), tsDataType);
+ }
+ nextOverlappedPageData.resetBatchData();
+ }
+ }
+ }
Review comment:
You can use the chunk and page statistics if possible. The code will be like as:
```
while (seriesReader.hasNextChunk()) {
if (seriesReader.canUseCurrentChunkStatistics()) {
if (seriesReader.currentChunkStatistics().getEndTime() > maxTime) {
maxTime = seriesReader.currentChunkStatistics().getEndTime();
queryResult.setPairResult(maxTime, seriesReader.currentChunkStatistics().getLastValue(), tsDataType);
}
seriesReader.skipCurrentChunk();
continue;
}
while (seriesReader.hasNextPage()) {
if (seriesReader.canUseCurrentPageStatistics()) {
if (seriesReader.currentPageStatistics().getEndTime() > maxTime) {
maxTime = seriesReader.currentPageStatistics().getEndTime();
queryResult.setPairResult(maxTime, seriesReader.currentPageStatistics().getLastValue(), tsDataType);
}
seriesReader.skipCurrentPage();
continue;
}
// cal by page data
while (seriesReader.hasNextOverlappedPage()) {
BatchData nextOverlappedPageData = seriesReader.nextOverlappedPage();
int maxIndex = nextOverlappedPageData.length() - 1;
if (maxIndex < 0) {
continue;
}
long time = nextOverlappedPageData.getTimeByIndex(maxIndex);
if (time > maxTime) {
maxTime = time;
queryResult.setPairResult(maxTime, nextOverlappedPageData.getValueInTimestamp(time), tsDataType);
}
nextOverlappedPageData.resetBatchData();
}
}
}
```
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] qiaojialin commented on a change in pull request
#821: [IOTDB-298]Last time-value query
Posted by GitBox <gi...@apache.org>.
qiaojialin commented on a change in pull request #821: [IOTDB-298]Last time-value query
URL: https://github.com/apache/incubator-iotdb/pull/821#discussion_r382381206
##########
File path: server/src/main/java/org/apache/iotdb/db/metadata/MNode.java
##########
@@ -238,4 +241,23 @@ public String getFullPath() {
}
return fullPath = builder.toString();
}
+
+ public TimeValuePair getCachedLast() {
+ return cachedLastValuePair;
+ }
+
+ public void setCachedLast(TimeValuePair timeValuePair) {
+ cachedLastValuePair = timeValuePair;
+ }
+
+ public void updateCachedLast(TimeValuePair timeValuePair) {
+ if (timeValuePair == null)
+ return;
+ if (cachedLastValuePair == null) {
+ cachedLastValuePair = new TimeValuePair(timeValuePair.getTimestamp(), timeValuePair.getValue());
+ } else if (timeValuePair.getTimestamp() > cachedLastValuePair.getTimestamp()) {
Review comment:
```suggestion
} else if (timeValuePair.getTimestamp() >= cachedLastValuePair.getTimestamp()) {
```
insert time-value 1-1
insert time-value 1-2
the vlaue of 1 should be updated to 2
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] wshao08 commented on a change in pull request
#821: [IOTDB-298]Last time-value query
Posted by GitBox <gi...@apache.org>.
wshao08 commented on a change in pull request #821: [IOTDB-298]Last time-value query
URL: https://github.com/apache/incubator-iotdb/pull/821#discussion_r385593366
##########
File path: docs/Documentation-CHN/UserGuide/5-Operation Manual/2-DML (Data Manipulation Language).md
##########
@@ -257,6 +257,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
Review comment:
fixed
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] qiaojialin commented on a change in pull request
#821: [IOTDB-298]Last time-value query
Posted by GitBox <gi...@apache.org>.
qiaojialin commented on a change in pull request #821: [IOTDB-298]Last time-value query
URL: https://github.com/apache/incubator-iotdb/pull/821#discussion_r384934300
##########
File path: server/src/main/java/org/apache/iotdb/db/service/TSServiceImpl.java
##########
@@ -646,7 +644,7 @@ private TSExecuteStatementResp executeQueryStatement(
resp.setOperationType(plan.getOperatorType().toString());
// generate the queryId for the operation
long queryId = generateQueryId(true);
- // put it into the corresponding Set
+ // put it into the corresponding SetO
Review comment:
？
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] wshao08 commented on a change in pull request
#821: [IOTDB-298]Last time-value query
Posted by GitBox <gi...@apache.org>.
wshao08 commented on a change in pull request #821: [IOTDB-298]Last time-value query
URL: https://github.com/apache/incubator-iotdb/pull/821#discussion_r386032592
##########
File path: server/src/main/java/org/apache/iotdb/db/query/executor/LastQueryExecutor.java
##########
@@ -0,0 +1,202 @@
+/*
+ * 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.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);
+
+ for (int i = seqFileResources.size() - 1; i >= 0; i--) {
+ List<ChunkMetaData> chunkMetadata =
+ loadSatisfiedChunkMetadata(seqFileResources.get(i), seriesPath, context);
+ if (!chunkMetadata.isEmpty()) {
+ ChunkMetaData lastChunkMetaData = chunkMetadata.get(chunkMetadata.size() - 1);
+ Statistics chunkStatistics = lastChunkMetaData.getStatistics();
+ resultPair = constructLastPair(
+ chunkStatistics.getEndTime(), chunkStatistics.getLastValue(), tsDataType);
+ break;
+ }
+ }
+
+ long version = 0;
+ for (TsFileResource resource : unseqFileResources) {
+ if (resource.getEndTimeMap().get(seriesPath.getDevice()) < resultPair.getTimestamp()) {
+ break;
+ }
+ List<ChunkMetaData> chunkMetadata = loadSatisfiedChunkMetadata(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 List<ChunkMetaData> loadSatisfiedChunkMetadata(
Review comment:
Fixed
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] qiaojialin commented on a change in pull request
#821: [IOTDB-298]Last time-value query
Posted by GitBox <gi...@apache.org>.
qiaojialin commented on a change in pull request #821: [IOTDB-298]Last time-value query
URL: https://github.com/apache/incubator-iotdb/pull/821#discussion_r382476404
##########
File path: server/src/main/java/org/apache/iotdb/db/query/executor/LastQueryExecutor.java
##########
@@ -0,0 +1,216 @@
+/*
+ * 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 org.apache.iotdb.db.exception.StorageEngineException;
+import org.apache.iotdb.db.exception.path.PathException;
+import org.apache.iotdb.db.exception.query.QueryProcessException;
+import org.apache.iotdb.db.metadata.MManager;
+import org.apache.iotdb.db.metadata.MNode;
+import org.apache.iotdb.db.qp.physical.crud.LastQueryPlan;
+import org.apache.iotdb.db.query.aggregation.AggregateResult;
+import org.apache.iotdb.db.query.context.QueryContext;
+import org.apache.iotdb.db.query.control.QueryResourceManager;
+import org.apache.iotdb.db.query.dataset.SingleDataSet;
+import org.apache.iotdb.db.query.reader.series.IAggregateReader;
+import org.apache.iotdb.db.query.reader.series.SeriesAggregateReader;
+import org.apache.iotdb.tsfile.exception.cache.CacheException;
+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.common.BatchData;
+import org.apache.iotdb.tsfile.read.common.Path;
+import org.apache.iotdb.tsfile.read.common.RowRecord;
+import org.apache.iotdb.tsfile.read.filter.basic.Filter;
+import org.apache.iotdb.tsfile.read.query.dataset.QueryDataSet;
+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 {
+
+ List<LastQueryResult> lastQueryResultList = new ArrayList<>();
+ for (int i = 0; i < selectedSeries.size(); i++) {
+ LastQueryResult lastQueryResult = calculateLastPairForOneSeries(selectedSeries.get(i), dataTypes.get(i), context);
+ lastQueryResultList.add(lastQueryResult);
+ }
+
+ RowRecord resultRecord = constructLastRowRecord(lastQueryResultList);
+ SingleDataSet dataSet = new SingleDataSet(selectedSeries, dataTypes);
+ dataSet.setRecord(resultRecord);
+ return dataSet;
+ }
+
+ /**
+ * get aggregation result for one series
+ *
+ * @param context query context
+ * @return AggregateResult list
+ */
+ private LastQueryResult calculateLastPairForOneSeries(
+ Path seriesPath, TSDataType tsDataType,
+ QueryContext context)
+ throws IOException, QueryProcessException, StorageEngineException {
+ LastQueryResult queryResult = new LastQueryResult();
+ MNode node = null;
+ try {
+ node = MManager.getInstance().getNodeByPathFromCache(seriesPath.toString());
+ } catch (PathException e) {
+ throw new QueryProcessException(e);
+ } catch (CacheException e) {
+ throw new QueryProcessException(e.getMessage());
+ }
+ if (node.getCachedLast() != null) {
+ queryResult.setPairResult(node.getCachedLast());
+ return queryResult;
+ }
+
+ // construct series reader without value filter
+ Filter timeFilter = null;
Review comment:
remove this field and use null directly
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] qiaojialin commented on a change in pull request
#821: [IOTDB-298]Last time-value query
Posted by GitBox <gi...@apache.org>.
qiaojialin commented on a change in pull request #821: [IOTDB-298]Last time-value query
URL: https://github.com/apache/incubator-iotdb/pull/821#discussion_r385673528
##########
File path: server/src/main/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessor.java
##########
@@ -569,18 +607,40 @@ public void insert(InsertPlan insertPlan) throws QueryProcessException {
}
}
+ long latestFlushedTime = globalLatestFlushedTimeForEachDevice.computeIfAbsent(
+ batchInsertPlan.getDeviceId(), k -> Long.MIN_VALUE);
// do not forget last part
if (before < loc) {
insertBatchToTsFileProcessor(batchInsertPlan, before, loc, isSequence, results,
beforeTimePartition);
}
+ updateBatchInsertPlanLast(batchInsertPlan, latestFlushedTime);
+ if (latestFlushedTime < batchInsertPlan.getMaxTime())
+ globalLatestFlushedTimeForEachDevice.put(batchInsertPlan.getDeviceId(), batchInsertPlan.getMaxTime());
Review comment:
```suggestion
if (latestFlushedTime < batchInsertPlan.getMaxTime()) {
globalLatestFlushedTimeForEachDevice.put(batchInsertPlan.getDeviceId(), batchInsertPlan.getMaxTime());
}
```
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] qiaojialin commented on a change in pull request
#821: [IOTDB-298]Last time-value query
Posted by GitBox <gi...@apache.org>.
qiaojialin commented on a change in pull request #821: [IOTDB-298]Last time-value query
URL: https://github.com/apache/incubator-iotdb/pull/821#discussion_r384929919
##########
File path: docs/Documentation-CHN/UserGuide/5-Operation Manual/2-DML (Data Manipulation Language).md
##########
@@ -257,6 +257,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
Review comment:
check this
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] wshao08 commented on a change in pull request
#821: [IOTDB-298]Last time-value query
Posted by GitBox <gi...@apache.org>.
wshao08 commented on a change in pull request #821: [IOTDB-298]Last time-value query
URL: https://github.com/apache/incubator-iotdb/pull/821#discussion_r386032640
##########
File path: server/src/main/java/org/apache/iotdb/db/metadata/mnode/InternalMNode.java
##########
@@ -20,6 +20,8 @@
import java.util.LinkedHashMap;
import java.util.Map;
+
+import org.apache.iotdb.tsfile.read.TimeValuePair;
Review comment:
Done
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] qiaojialin commented on a change in pull request
#821: [IOTDB-298]Last time-value query
Posted by GitBox <gi...@apache.org>.
qiaojialin commented on a change in pull request #821: [IOTDB-298]Last time-value query
URL: https://github.com/apache/incubator-iotdb/pull/821#discussion_r385675337
##########
File path: server/src/main/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessor.java
##########
@@ -1695,6 +1759,10 @@ private void updateLatestTimeMap(TsFileResource newTsFileResource) {
.computeIfAbsent(timePartitionId, id -> new HashMap<String, Long>())
.put(device, endTime);
}
+ if (!globalLatestFlushedTimeForEachDevice.containsKey(device)
Review comment:
update javadoc
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] qiaojialin merged pull request #821:
[IOTDB-298]Last time-value query
Posted by GitBox <gi...@apache.org>.
qiaojialin merged pull request #821: [IOTDB-298]Last time-value query
URL: https://github.com/apache/incubator-iotdb/pull/821
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] qiaojialin commented on a change in pull request
#821: [IOTDB-298]Last time-value query
Posted by GitBox <gi...@apache.org>.
qiaojialin commented on a change in pull request #821: [IOTDB-298]Last time-value query
URL: https://github.com/apache/incubator-iotdb/pull/821#discussion_r385680971
##########
File path: server/src/main/java/org/apache/iotdb/db/query/executor/LastQueryExecutor.java
##########
@@ -0,0 +1,202 @@
+/*
+ * 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.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);
+
+ for (int i = seqFileResources.size() - 1; i >= 0; i--) {
Review comment:
no need to loop, get the last one directly
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] wshao08 commented on a change in pull request
#821: [IOTDB-298]Last time-value query
Posted by GitBox <gi...@apache.org>.
wshao08 commented on a change in pull request #821: [IOTDB-298]Last time-value query
URL: https://github.com/apache/incubator-iotdb/pull/821#discussion_r386032645
##########
File path: server/src/main/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessor.java
##########
@@ -569,18 +607,40 @@ public void insert(InsertPlan insertPlan) throws QueryProcessException {
}
}
+ long latestFlushedTime = globalLatestFlushedTimeForEachDevice.computeIfAbsent(
+ batchInsertPlan.getDeviceId(), k -> Long.MIN_VALUE);
// do not forget last part
if (before < loc) {
insertBatchToTsFileProcessor(batchInsertPlan, before, loc, isSequence, results,
beforeTimePartition);
}
+ updateBatchInsertPlanLast(batchInsertPlan, latestFlushedTime);
+ if (latestFlushedTime < batchInsertPlan.getMaxTime())
+ globalLatestFlushedTimeForEachDevice.put(batchInsertPlan.getDeviceId(), batchInsertPlan.getMaxTime());
return results;
} finally {
writeUnlock();
}
}
+ public void updateBatchInsertPlanLast(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(
Review comment:
Fixed
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] qiaojialin commented on a change in pull request
#821: [IOTDB-298]Last time-value query
Posted by GitBox <gi...@apache.org>.
qiaojialin commented on a change in pull request #821: [IOTDB-298]Last time-value query
URL: https://github.com/apache/incubator-iotdb/pull/821#discussion_r385676055
##########
File path: server/src/main/java/org/apache/iotdb/db/metadata/mnode/LeafMNode.java
##########
@@ -74,4 +77,25 @@ public int getLeafCount() {
public MeasurementSchema getSchema() {
return schema;
}
+
+ public TimeValuePair getCachedLast() {
+ return cachedLastValuePair;
+ }
+
+ public synchronized void updateCachedLast(TimeValuePair timeValuePair, boolean insertionUpdate,
Review comment:
```suggestion
public synchronized void updateCachedLast(TimeValuePair timeValuePair, boolean updateAtSameTime,
```
add java doc
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] qiaojialin commented on a change in pull request
#821: [IOTDB-298]Last time-value query
Posted by GitBox <gi...@apache.org>.
qiaojialin commented on a change in pull request #821: [IOTDB-298]Last time-value query
URL: https://github.com/apache/incubator-iotdb/pull/821#discussion_r382475822
##########
File path: server/src/main/java/org/apache/iotdb/db/qp/executor/PlanExecutor.java
##########
@@ -772,8 +779,12 @@ private void checkPathExists(MNode node, String fullPath, MeasurementSchema sche
measurementNode.getSchema().getType()));
}
}
- return storageEngine.insertBatch(batchInsertPlan);
-
+ Integer[] results = storageEngine.insertBatch(batchInsertPlan);
+ for (int i = 0; i < measurementList.length; i++) {
+ MNode measurementNode = node.getChild(measurementList[i]);
+ measurementNode.updateCachedLast(batchInsertPlan.composeLastTimeValuePair(i));
Review comment:
Please check this condition:
insert into two INT32 series s1, s2 with values 1, "b".
These two values will all be cached in the MNode.
However, when inserting to the memtable, the inserting "b" to s2 will throw RuntimeException.
You'd better check this value type before adding it to cache. (The MNode is classified into three type in mmanager_refactor branch, the measurement node is a LeafNode, you could merge this branch)
Refer to AbstractMemtable.insert()
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] qiaojialin commented on a change in pull request
#821: [IOTDB-298]Last time-value query
Posted by GitBox <gi...@apache.org>.
qiaojialin commented on a change in pull request #821: [IOTDB-298]Last time-value query
URL: https://github.com/apache/incubator-iotdb/pull/821#discussion_r385683715
##########
File path: server/src/main/java/org/apache/iotdb/db/query/executor/LastQueryExecutor.java
##########
@@ -0,0 +1,202 @@
+/*
+ * 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.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);
Review comment:
```suggestion
TimeValuePair resultPair;
```
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] qiaojialin commented on a change in pull request
#821: [IOTDB-298]Last time-value query
Posted by GitBox <gi...@apache.org>.
qiaojialin commented on a change in pull request #821: [IOTDB-298]Last time-value query
URL: https://github.com/apache/incubator-iotdb/pull/821#discussion_r385674357
##########
File path: server/src/main/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessor.java
##########
@@ -569,18 +607,40 @@ public void insert(InsertPlan insertPlan) throws QueryProcessException {
}
}
+ long latestFlushedTime = globalLatestFlushedTimeForEachDevice.computeIfAbsent(
+ batchInsertPlan.getDeviceId(), k -> Long.MIN_VALUE);
// do not forget last part
if (before < loc) {
insertBatchToTsFileProcessor(batchInsertPlan, before, loc, isSequence, results,
beforeTimePartition);
}
+ updateBatchInsertPlanLast(batchInsertPlan, latestFlushedTime);
+ if (latestFlushedTime < batchInsertPlan.getMaxTime())
+ globalLatestFlushedTimeForEachDevice.put(batchInsertPlan.getDeviceId(), batchInsertPlan.getMaxTime());
return results;
} finally {
writeUnlock();
}
}
+ public void updateBatchInsertPlanLast(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(
Review comment:
```suggestion
((LeafMNode) measurementNode).updateCachedLast(
```
import java-google-code.xml code style and format the code
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] JackieTien97 commented on a change in pull
request #821: [IOTDB-298]Last time-value query
Posted by GitBox <gi...@apache.org>.
JackieTien97 commented on a change in pull request #821: [IOTDB-298]Last time-value query
URL: https://github.com/apache/incubator-iotdb/pull/821#discussion_r380538229
##########
File path: server/src/main/java/org/apache/iotdb/db/qp/executor/PlanExecutor.java
##########
@@ -772,8 +767,9 @@ private void checkPathExists(MNode node, String fullPath, MeasurementSchema sche
measurementNode.getSchema().getType()));
}
}
- return storageEngine.insertBatch(batchInsertPlan);
-
+ Integer[] results = storageEngine.insertBatch(batchInsertPlan);
+ batchInsertPlan.updateMNodeLastValues(node);
Review comment:
The same as before
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] qiaojialin commented on a change in pull request
#821: [IOTDB-298]Last time-value query
Posted by GitBox <gi...@apache.org>.
qiaojialin commented on a change in pull request #821: [IOTDB-298]Last time-value query
URL: https://github.com/apache/incubator-iotdb/pull/821#discussion_r384931262
##########
File path: server/src/main/java/org/apache/iotdb/db/qp/logical/crud/SelectOperator.java
##########
@@ -30,6 +30,7 @@
private List<Path> suffixList;
private List<String> aggregations;
+ private boolean hasLast;
Review comment:
```suggestion
private boolean lastQuery;
```
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] qiaojialin commented on a change in pull request
#821: [IOTDB-298]Last time-value query
Posted by GitBox <gi...@apache.org>.
qiaojialin commented on a change in pull request #821: [IOTDB-298]Last time-value query
URL: https://github.com/apache/incubator-iotdb/pull/821#discussion_r384930578
##########
File path: server/src/main/java/org/apache/iotdb/db/metadata/mnode/MNode.java
##########
@@ -43,6 +44,8 @@
*/
protected String fullPath;
+ private TimeValuePair cachedLastValuePair = null;
Review comment:
move this to LeafMNode
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] qiaojialin commented on a change in pull request
#821: [IOTDB-298]Last time-value query
Posted by GitBox <gi...@apache.org>.
qiaojialin commented on a change in pull request #821: [IOTDB-298]Last time-value query
URL: https://github.com/apache/incubator-iotdb/pull/821#discussion_r382475822
##########
File path: server/src/main/java/org/apache/iotdb/db/qp/executor/PlanExecutor.java
##########
@@ -772,8 +779,12 @@ private void checkPathExists(MNode node, String fullPath, MeasurementSchema sche
measurementNode.getSchema().getType()));
}
}
- return storageEngine.insertBatch(batchInsertPlan);
-
+ Integer[] results = storageEngine.insertBatch(batchInsertPlan);
+ for (int i = 0; i < measurementList.length; i++) {
+ MNode measurementNode = node.getChild(measurementList[i]);
+ measurementNode.updateCachedLast(batchInsertPlan.composeLastTimeValuePair(i));
Review comment:
Please check this condition:
insert into two INT32 series s1, s2 with values 1, "b".
These two values will all be cached in the MNode.
However, when inserting to the memtable, the inserting "b" to s2 will throw RuntimeException.
You'd better check this value type before adding it to cache. (The MNode is classified into three type in mmanager_refactor branch, the measurement node is a LeafNode, you could merge this branch)
Refer to AbstractMemtable.insert()
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] wshao08 commented on a change in pull request
#821: [IOTDB-298]Last time-value query
Posted by GitBox <gi...@apache.org>.
wshao08 commented on a change in pull request #821: [IOTDB-298]Last time-value query
URL: https://github.com/apache/incubator-iotdb/pull/821#discussion_r386032658
##########
File path: server/src/main/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessor.java
##########
@@ -569,18 +607,40 @@ public void insert(InsertPlan insertPlan) throws QueryProcessException {
}
}
+ long latestFlushedTime = globalLatestFlushedTimeForEachDevice.computeIfAbsent(
+ batchInsertPlan.getDeviceId(), k -> Long.MIN_VALUE);
// do not forget last part
if (before < loc) {
insertBatchToTsFileProcessor(batchInsertPlan, before, loc, isSequence, results,
beforeTimePartition);
}
+ updateBatchInsertPlanLast(batchInsertPlan, latestFlushedTime);
+ if (latestFlushedTime < batchInsertPlan.getMaxTime())
+ globalLatestFlushedTimeForEachDevice.put(batchInsertPlan.getDeviceId(), batchInsertPlan.getMaxTime());
Review comment:
Moved
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] wshao08 commented on a change in pull request
#821: [IOTDB-298]Last time-value query
Posted by GitBox <gi...@apache.org>.
wshao08 commented on a change in pull request #821: [IOTDB-298]Last time-value query
URL: https://github.com/apache/incubator-iotdb/pull/821#discussion_r384646004
##########
File path: server/src/main/java/org/apache/iotdb/db/qp/physical/crud/BatchInsertPlan.java
##########
@@ -288,6 +291,19 @@ public long getMaxTime() {
return maxTime;
}
+ public TimeValuePair composeLastTimeValuePair(int measurementIndex) {
+ long maxTime = Long.MIN_VALUE;
+ int maxIndex = 0;
+ for (int i = 0; i < times.length; i++) {
Review comment:
Done
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] wshao08 commented on a change in pull request
#821: [IOTDB-298]Last time-value query
Posted by GitBox <gi...@apache.org>.
wshao08 commented on a change in pull request #821: [IOTDB-298]Last time-value query
URL: https://github.com/apache/incubator-iotdb/pull/821#discussion_r380561106
##########
File path: server/src/main/java/org/apache/iotdb/db/metadata/MNode.java
##########
@@ -22,10 +22,15 @@
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.Map;
+
+import com.sun.rowset.internal.Row;
Review comment:
Removed
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] qiaojialin commented on a change in pull request
#821: [IOTDB-298]Last time-value query
Posted by GitBox <gi...@apache.org>.
qiaojialin commented on a change in pull request #821: [IOTDB-298]Last time-value query
URL: https://github.com/apache/incubator-iotdb/pull/821#discussion_r385667354
##########
File path: server/src/main/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessor.java
##########
@@ -496,15 +511,38 @@ public void insert(InsertPlan insertPlan) throws QueryProcessException {
latestFlushedTimeForEachDevice.computeIfAbsent(timePartitionId, id -> new HashMap<>())
.putIfAbsent(insertPlan.getDeviceId(), Long.MIN_VALUE);
+ long latestFlushedTime = globalLatestFlushedTimeForEachDevice.computeIfAbsent(
+ insertPlan.getDeviceId(), k -> Long.MIN_VALUE);
// insert to sequence or unSequence file
insertToTsFileProcessor(insertPlan,
insertPlan.getTime() > latestFlushedTimeForEachDevice.get(timePartitionId)
.get(insertPlan.getDeviceId()));
+
+ updateInsertPlanLast(insertPlan, latestFlushedTime);
+ if (latestFlushedTime < insertPlan.getTime())
+ globalLatestFlushedTimeForEachDevice.put(insertPlan.getDeviceId(), insertPlan.getTime());
Review comment:
```suggestion
if (latestFlushedTime < insertPlan.getTime()) {
globalLatestFlushedTimeForEachDevice.put(insertPlan.getDeviceId(), insertPlan.getTime());
}
```
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] qiaojialin commented on a change in pull request
#821: [IOTDB-298]Last time-value query
Posted by GitBox <gi...@apache.org>.
qiaojialin commented on a change in pull request #821: [IOTDB-298]Last time-value query
URL: https://github.com/apache/incubator-iotdb/pull/821#discussion_r382465639
##########
File path: server/src/main/java/org/apache/iotdb/db/query/executor/LastQueryExecutor.java
##########
@@ -0,0 +1,216 @@
+/*
+ * 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 org.apache.iotdb.db.exception.StorageEngineException;
+import org.apache.iotdb.db.exception.path.PathException;
+import org.apache.iotdb.db.exception.query.QueryProcessException;
+import org.apache.iotdb.db.metadata.MManager;
+import org.apache.iotdb.db.metadata.MNode;
+import org.apache.iotdb.db.qp.physical.crud.LastQueryPlan;
+import org.apache.iotdb.db.query.aggregation.AggregateResult;
+import org.apache.iotdb.db.query.context.QueryContext;
+import org.apache.iotdb.db.query.control.QueryResourceManager;
+import org.apache.iotdb.db.query.dataset.SingleDataSet;
+import org.apache.iotdb.db.query.reader.series.IAggregateReader;
+import org.apache.iotdb.db.query.reader.series.SeriesAggregateReader;
+import org.apache.iotdb.tsfile.exception.cache.CacheException;
+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.common.BatchData;
+import org.apache.iotdb.tsfile.read.common.Path;
+import org.apache.iotdb.tsfile.read.common.RowRecord;
+import org.apache.iotdb.tsfile.read.filter.basic.Filter;
+import org.apache.iotdb.tsfile.read.query.dataset.QueryDataSet;
+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 {
+
+ List<LastQueryResult> lastQueryResultList = new ArrayList<>();
+ for (int i = 0; i < selectedSeries.size(); i++) {
+ LastQueryResult lastQueryResult = calculateLastPairForOneSeries(selectedSeries.get(i), dataTypes.get(i), context);
+ lastQueryResultList.add(lastQueryResult);
+ }
+
+ RowRecord resultRecord = constructLastRowRecord(lastQueryResultList);
Review comment:
The result format should be updated to path, time, value
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] qiaojialin commented on a change in pull request
#821: [IOTDB-298]Last time-value query
Posted by GitBox <gi...@apache.org>.
qiaojialin commented on a change in pull request #821: [IOTDB-298]Last time-value query
URL: https://github.com/apache/incubator-iotdb/pull/821#discussion_r385678955
##########
File path: server/src/main/java/org/apache/iotdb/db/metadata/mnode/LeafMNode.java
##########
@@ -74,4 +77,25 @@ public int getLeafCount() {
public MeasurementSchema getSchema() {
return schema;
}
+
+ public TimeValuePair getCachedLast() {
+ return cachedLastValuePair;
+ }
+
+ public synchronized void updateCachedLast(TimeValuePair timeValuePair, boolean insertionUpdate,
+ Long latestFlushedTime) {
+ if (timeValuePair == null || timeValuePair.getValue() == null) return;
Review comment:
format
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] wshao08 commented on a change in pull request
#821: [IOTDB-298]Last time-value query
Posted by GitBox <gi...@apache.org>.
wshao08 commented on a change in pull request #821: [IOTDB-298]Last time-value query
URL: https://github.com/apache/incubator-iotdb/pull/821#discussion_r386032609
##########
File path: server/src/main/java/org/apache/iotdb/db/query/executor/LastQueryExecutor.java
##########
@@ -0,0 +1,202 @@
+/*
+ * 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.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);
+
+ for (int i = seqFileResources.size() - 1; i >= 0; i--) {
+ List<ChunkMetaData> chunkMetadata =
+ loadSatisfiedChunkMetadata(seqFileResources.get(i), seriesPath, context);
+ if (!chunkMetadata.isEmpty()) {
+ ChunkMetaData lastChunkMetaData = chunkMetadata.get(chunkMetadata.size() - 1);
+ Statistics chunkStatistics = lastChunkMetaData.getStatistics();
+ resultPair = constructLastPair(
+ chunkStatistics.getEndTime(), chunkStatistics.getLastValue(), tsDataType);
+ break;
+ }
+ }
+
+ long version = 0;
+ for (TsFileResource resource : unseqFileResources) {
+ if (resource.getEndTimeMap().get(seriesPath.getDevice()) < resultPair.getTimestamp()) {
+ break;
+ }
+ List<ChunkMetaData> chunkMetadata = loadSatisfiedChunkMetadata(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 List<ChunkMetaData> loadSatisfiedChunkMetadata(
+ 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) {
+ if (data.getChunkLoader() == null) {
+ 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;
+ }
+
+ private TimeValuePair constructLastPair(long timestamp, Object value, TSDataType dataType) {
+ TimeValuePair lastPair = new TimeValuePair(timestamp, TsPrimitiveType.getByType(dataType, value));
+ return lastPair;
Review comment:
Fixed
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] wshao08 opened a new pull request #821:
[IOTDB-298]Last time-value query
Posted by GitBox <gi...@apache.org>.
wshao08 opened a new pull request #821: [IOTDB-298]Last time-value query
URL: https://github.com/apache/incubator-iotdb/pull/821
This PR implements a new LAST query for IoTDB
The LAST query is to return the most recent value of the given timeseries in a time-value format.
The basic Sql syntax of Last query is:
```
select last(path) from <prefixPath>
```
In order to trace the latest timestamp and value as quick as possible, this PR adds a cache to store this latest time and value in MNodes. This cache will be updated when values with larger timestamps are written to the specific timeseries.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] wshao08 commented on a change in pull request
#821: [IOTDB-298]Last time-value query
Posted by GitBox <gi...@apache.org>.
wshao08 commented on a change in pull request #821: [IOTDB-298]Last time-value query
URL: https://github.com/apache/incubator-iotdb/pull/821#discussion_r384645843
##########
File path: server/src/main/java/org/apache/iotdb/db/metadata/MNode.java
##########
@@ -238,4 +241,23 @@ public String getFullPath() {
}
return fullPath = builder.toString();
}
+
+ public TimeValuePair getCachedLast() {
+ return cachedLastValuePair;
+ }
+
+ public void setCachedLast(TimeValuePair timeValuePair) {
+ cachedLastValuePair = timeValuePair;
+ }
+
+ public void updateCachedLast(TimeValuePair timeValuePair) {
+ if (timeValuePair == null)
+ return;
+ if (cachedLastValuePair == null) {
+ cachedLastValuePair = new TimeValuePair(timeValuePair.getTimestamp(), timeValuePair.getValue());
+ } else if (timeValuePair.getTimestamp() > cachedLastValuePair.getTimestamp()) {
Review comment:
Fixed
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] JackieTien97 commented on a change in pull
request #821: [IOTDB-298]Last time-value query
Posted by GitBox <gi...@apache.org>.
JackieTien97 commented on a change in pull request #821: [IOTDB-298]Last time-value query
URL: https://github.com/apache/incubator-iotdb/pull/821#discussion_r386360706
##########
File path: server/src/main/java/org/apache/iotdb/db/utils/FileLoaderUtils.java
##########
@@ -64,4 +72,40 @@ public static void updateTsFileResource(TsFileMetaData metaData, TsFileSequenceR
}
}
}
+
+ 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) {
+ if (data.getChunkLoader() == null) {
+ TsFileSequenceReader tsFileSequenceReader =
+ FileReaderManager.getInstance().get(resource, resource.isClosed());
+ data.setChunkLoader(new DiskChunkLoader(tsFileSequenceReader));
+ }
Review comment:
Remove the if judgement, it will cause the file closed null pointer exception.
like:
for (ChunkMetaData data : currentChunkMetaDataList) {
TsFileSequenceReader tsFileSequenceReader =
FileReaderManager.getInstance().get(resource, resource.isClosed());
data.setChunkLoader(new DiskChunkLoader(tsFileSequenceReader));
}
it's ok to do so. because tsFileSequenceReader has been cached.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] qiaojialin commented on a change in pull request
#821: [IOTDB-298]Last time-value query
Posted by GitBox <gi...@apache.org>.
qiaojialin commented on a change in pull request #821: [IOTDB-298]Last time-value query
URL: https://github.com/apache/incubator-iotdb/pull/821#discussion_r382482041
##########
File path: server/src/main/java/org/apache/iotdb/db/query/executor/LastQueryExecutor.java
##########
@@ -0,0 +1,216 @@
+/*
+ * 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 org.apache.iotdb.db.exception.StorageEngineException;
+import org.apache.iotdb.db.exception.path.PathException;
+import org.apache.iotdb.db.exception.query.QueryProcessException;
+import org.apache.iotdb.db.metadata.MManager;
+import org.apache.iotdb.db.metadata.MNode;
+import org.apache.iotdb.db.qp.physical.crud.LastQueryPlan;
+import org.apache.iotdb.db.query.aggregation.AggregateResult;
+import org.apache.iotdb.db.query.context.QueryContext;
+import org.apache.iotdb.db.query.control.QueryResourceManager;
+import org.apache.iotdb.db.query.dataset.SingleDataSet;
+import org.apache.iotdb.db.query.reader.series.IAggregateReader;
+import org.apache.iotdb.db.query.reader.series.SeriesAggregateReader;
+import org.apache.iotdb.tsfile.exception.cache.CacheException;
+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.common.BatchData;
+import org.apache.iotdb.tsfile.read.common.Path;
+import org.apache.iotdb.tsfile.read.common.RowRecord;
+import org.apache.iotdb.tsfile.read.filter.basic.Filter;
+import org.apache.iotdb.tsfile.read.query.dataset.QueryDataSet;
+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 {
+
+ List<LastQueryResult> lastQueryResultList = new ArrayList<>();
+ for (int i = 0; i < selectedSeries.size(); i++) {
+ LastQueryResult lastQueryResult = calculateLastPairForOneSeries(selectedSeries.get(i), dataTypes.get(i), context);
+ lastQueryResultList.add(lastQueryResult);
+ }
+
+ RowRecord resultRecord = constructLastRowRecord(lastQueryResultList);
+ SingleDataSet dataSet = new SingleDataSet(selectedSeries, dataTypes);
+ dataSet.setRecord(resultRecord);
+ return dataSet;
+ }
+
+ /**
+ * get aggregation result for one series
+ *
+ * @param context query context
+ * @return AggregateResult list
+ */
+ private LastQueryResult calculateLastPairForOneSeries(
+ Path seriesPath, TSDataType tsDataType,
+ QueryContext context)
+ throws IOException, QueryProcessException, StorageEngineException {
+ LastQueryResult queryResult = new LastQueryResult();
+ MNode node = null;
+ try {
+ node = MManager.getInstance().getNodeByPathFromCache(seriesPath.toString());
+ } catch (PathException e) {
+ throw new QueryProcessException(e);
+ } catch (CacheException e) {
+ throw new QueryProcessException(e.getMessage());
+ }
+ if (node.getCachedLast() != null) {
+ queryResult.setPairResult(node.getCachedLast());
+ return queryResult;
+ }
+
+ // construct series reader without value filter
+ Filter timeFilter = null;
+ IAggregateReader seriesReader = new SeriesAggregateReader(
+ seriesPath, tsDataType, context, QueryResourceManager.getInstance()
+ .getQueryDataSource(seriesPath, context, timeFilter), timeFilter, null);
+
+ long maxTime = Long.MIN_VALUE;
+ while (seriesReader.hasNextChunk()) {
+ // cal by chunk statistics
+ if (seriesReader.canUseCurrentChunkStatistics()) {
+ Statistics chunkStatistics = seriesReader.currentChunkStatistics();
+ if (chunkStatistics.getEndTime() > maxTime) {
+ maxTime = chunkStatistics.getEndTime();
+ queryResult.setPairResult(maxTime, chunkStatistics.getLastValue(), tsDataType);
+ }
+ seriesReader.skipCurrentChunk();
+ continue;
+ }
+ while (seriesReader.hasNextPage()) {
+ //cal by page statistics
+ if (seriesReader.canUseCurrentPageStatistics()) {
+ Statistics pageStatistic = seriesReader.currentPageStatistics();
+ if (pageStatistic.getEndTime() > maxTime) {
+ maxTime = pageStatistic.getEndTime();
+ queryResult.setPairResult(maxTime, pageStatistic.getLastValue(), tsDataType);
+ }
+ seriesReader.skipCurrentPage();
+ continue;
+ }
+ // cal by page data
+ while (seriesReader.hasNextOverlappedPage()) {
+ BatchData nextOverlappedPageData = seriesReader.nextOverlappedPage();
+ int maxIndex = nextOverlappedPageData.length() - 1;
+ if (maxIndex < 0) {
+ continue;
+ }
+ long time = nextOverlappedPageData.getTimeByIndex(maxIndex);
+ if (time > maxTime) {
+ maxTime = time;
+ queryResult.setPairResult(maxTime, nextOverlappedPageData.getValueInTimestamp(time), tsDataType);
+ }
+ nextOverlappedPageData.resetBatchData();
+ }
+ }
+ }
+ if (queryResult.hasResult())
+ node.setCachedLast(queryResult.getPairResult());
Review comment:
Notice: considering inserting a point while querying.
It's better to remove the setCachedLast and change updateCachedLast method to a synchronized one. Then, extend updateCachedLast with a parameter: boolean highPriority, which is to determine whether update the value if the time is equal to the previous cached one.
Then, in the normal inserting process, update the cache with high priority. Here, update the cache with low priority.
This could resolve the condition: there is an insert (10,11) when you get the last point (10,10) from disk.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] qiaojialin commented on a change in pull request
#821: [IOTDB-298]Last time-value query
Posted by GitBox <gi...@apache.org>.
qiaojialin commented on a change in pull request #821: [IOTDB-298]Last time-value query
URL: https://github.com/apache/incubator-iotdb/pull/821#discussion_r385661250
##########
File path: server/src/main/java/org/apache/iotdb/db/engine/StorageEngine.java
##########
@@ -235,7 +235,6 @@ public void insert(InsertPlan insertPlan)
throw new QueryProcessException(e);
}
}
-
Review comment:
add a blank line...
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] qiaojialin commented on a change in pull request
#821: [IOTDB-298]Last time-value query
Posted by GitBox <gi...@apache.org>.
qiaojialin commented on a change in pull request #821: [IOTDB-298]Last time-value query
URL: https://github.com/apache/incubator-iotdb/pull/821#discussion_r385672511
##########
File path: server/src/main/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessor.java
##########
@@ -496,15 +511,38 @@ public void insert(InsertPlan insertPlan) throws QueryProcessException {
latestFlushedTimeForEachDevice.computeIfAbsent(timePartitionId, id -> new HashMap<>())
.putIfAbsent(insertPlan.getDeviceId(), Long.MIN_VALUE);
+ long latestFlushedTime = globalLatestFlushedTimeForEachDevice.computeIfAbsent(
+ insertPlan.getDeviceId(), k -> Long.MIN_VALUE);
// insert to sequence or unSequence file
insertToTsFileProcessor(insertPlan,
insertPlan.getTime() > latestFlushedTimeForEachDevice.get(timePartitionId)
.get(insertPlan.getDeviceId()));
+
+ updateInsertPlanLast(insertPlan, latestFlushedTime);
+ if (latestFlushedTime < insertPlan.getTime())
+ globalLatestFlushedTimeForEachDevice.put(insertPlan.getDeviceId(), insertPlan.getTime());
Review comment:
Then, move these three lines to insertToTsFileProcessor(), make it updated at the same time as the lastFlushTimeForEachDevice
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] qiaojialin commented on a change in pull request
#821: [IOTDB-298]Last time-value query
Posted by GitBox <gi...@apache.org>.
qiaojialin commented on a change in pull request #821: [IOTDB-298]Last time-value query
URL: https://github.com/apache/incubator-iotdb/pull/821#discussion_r384931711
##########
File path: server/src/main/java/org/apache/iotdb/db/query/executor/LastQueryExecutor.java
##########
@@ -0,0 +1,172 @@
+/*
+ * 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.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.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.QueryResourceManager;
+import org.apache.iotdb.db.query.dataset.ListDataSet;
+import org.apache.iotdb.db.query.reader.series.IAggregateReader;
+import org.apache.iotdb.db.query.reader.series.SeriesAggregateReader;
+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.common.BatchData;
+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(
Review comment:
```suggestion
ListDataSet dataSet = new ListDataSet(
```
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] qiaojialin commented on a change in pull request
#821: [IOTDB-298]Last time-value query
Posted by GitBox <gi...@apache.org>.
qiaojialin commented on a change in pull request #821: [IOTDB-298]Last time-value query
URL: https://github.com/apache/incubator-iotdb/pull/821#discussion_r385673361
##########
File path: server/src/main/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessor.java
##########
@@ -569,18 +607,40 @@ public void insert(InsertPlan insertPlan) throws QueryProcessException {
}
}
+ long latestFlushedTime = globalLatestFlushedTimeForEachDevice.computeIfAbsent(
Review comment:
```suggestion
long globalLatestFlushedTime = globalLatestFlushedTimeForEachDevice.computeIfAbsent(
```
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] JackieTien97 commented on a change in pull
request #821: [IOTDB-298]Last time-value query
Posted by GitBox <gi...@apache.org>.
JackieTien97 commented on a change in pull request #821: [IOTDB-298]Last time-value query
URL: https://github.com/apache/incubator-iotdb/pull/821#discussion_r380532638
##########
File path: server/src/main/java/org/apache/iotdb/db/query/executor/LastQueryExecutor.java
##########
@@ -0,0 +1,201 @@
+/*
+ * 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 org.apache.iotdb.db.exception.StorageEngineException;
+import org.apache.iotdb.db.exception.path.PathException;
+import org.apache.iotdb.db.exception.query.QueryProcessException;
+import org.apache.iotdb.db.metadata.MManager;
+import org.apache.iotdb.db.metadata.MNode;
+import org.apache.iotdb.db.qp.constant.SQLConstant;
+import org.apache.iotdb.db.qp.physical.crud.AggregationPlan;
+import org.apache.iotdb.db.qp.physical.crud.LastQueryPlan;
+import org.apache.iotdb.db.query.aggregation.AggregateResult;
+import org.apache.iotdb.db.query.context.QueryContext;
+import org.apache.iotdb.db.query.control.QueryResourceManager;
+import org.apache.iotdb.db.query.dataset.SingleDataSet;
+import org.apache.iotdb.db.query.factory.AggreResultFactory;
+import org.apache.iotdb.db.query.reader.series.IAggregateReader;
+import org.apache.iotdb.db.query.reader.series.SeriesAggregateReader;
+import org.apache.iotdb.tsfile.exception.cache.CacheException;
+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.common.BatchData;
+import org.apache.iotdb.tsfile.read.common.Path;
+import org.apache.iotdb.tsfile.read.common.RowRecord;
+import org.apache.iotdb.tsfile.read.filter.basic.Filter;
+import org.apache.iotdb.tsfile.read.query.dataset.QueryDataSet;
+import org.apache.iotdb.tsfile.utils.TsPrimitiveType;
+import org.omg.CORBA.TIMEOUT;
Review comment:
A lot of imports are not used. You should check that and delete them
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] qiaojialin commented on a change in pull request
#821: [IOTDB-298]Last time-value query
Posted by GitBox <gi...@apache.org>.
qiaojialin commented on a change in pull request #821: [IOTDB-298]Last time-value query
URL: https://github.com/apache/incubator-iotdb/pull/821#discussion_r382464574
##########
File path: server/src/main/java/org/apache/iotdb/db/qp/strategy/LogicalGenerator.java
##########
@@ -1047,6 +1048,16 @@ public void enterSelectElement(SelectElementContext ctx) {
queryOp.setSelectOperator(selectOp);
}
+ @Override
+ public void enterLastElement(SqlBaseParser.LastElementContext ctx) {
+ super.enterLastElement(ctx);
+ selectOp = new SelectOperator(SQLConstant.TOK_SELECT);
+ LastFunctionCallContext functionCallContext = ctx.lastFunctionCall();
+ Path path = parseSuffixPath(functionCallContext.suffixPath());
+ selectOp.addLastPath(path, functionCallContext.LAST().getText());
+ queryOp.setSelectOperator(selectOp);
Review comment:
the sql needs update
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] qiaojialin commented on a change in pull request
#821: [IOTDB-298]Last time-value query
Posted by GitBox <gi...@apache.org>.
qiaojialin commented on a change in pull request #821: [IOTDB-298]Last time-value query
URL: https://github.com/apache/incubator-iotdb/pull/821#discussion_r385678908
##########
File path: server/src/main/java/org/apache/iotdb/db/metadata/mnode/LeafMNode.java
##########
@@ -74,4 +77,25 @@ public int getLeafCount() {
public MeasurementSchema getSchema() {
return schema;
}
+
+ public TimeValuePair getCachedLast() {
+ return cachedLastValuePair;
+ }
+
+ public synchronized void updateCachedLast(TimeValuePair timeValuePair, boolean insertionUpdate,
+ Long latestFlushedTime) {
+ if (timeValuePair == null || timeValuePair.getValue() == null) return;
+
+ if (cachedLastValuePair == null){
+ // If no cached last, a last read or an insertion to a sequenceTsFile will update cache.
Review comment:
```suggestion
// If no cached last, (1) a last query or (2) an unseq insertion at globalLastFlushTime or (3) a seq insertion will update cache.
```
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] wshao08 commented on a change in pull request
#821: [IOTDB-298]Last time-value query
Posted by GitBox <gi...@apache.org>.
wshao08 commented on a change in pull request #821: [IOTDB-298]Last time-value query
URL: https://github.com/apache/incubator-iotdb/pull/821#discussion_r384918260
##########
File path: server/src/main/java/org/apache/iotdb/db/qp/physical/crud/BatchInsertPlan.java
##########
@@ -288,6 +291,19 @@ public long getMaxTime() {
return maxTime;
}
+ public TimeValuePair composeLastTimeValuePair(int measurementIndex) {
+ long maxTime = Long.MIN_VALUE;
+ int maxIndex = 0;
+ for (int i = 0; i < times.length; i++) {
+ if (times[i] > maxTime) {
+ maxTime = times[i];
+ maxIndex = i;
+ }
+ }
Review comment:
Fixed
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] JackieTien97 commented on a change in pull
request #821: [IOTDB-298]Last time-value query
Posted by GitBox <gi...@apache.org>.
JackieTien97 commented on a change in pull request #821: [IOTDB-298]Last time-value query
URL: https://github.com/apache/incubator-iotdb/pull/821#discussion_r380518901
##########
File path: server/src/main/java/org/apache/iotdb/db/metadata/MNode.java
##########
@@ -22,10 +22,15 @@
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.Map;
+
+import com.sun.rowset.internal.Row;
import org.apache.iotdb.db.conf.IoTDBConstant;
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.read.common.RowRecord;
Review comment:
this import is not used, please remove it.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] JackieTien97 commented on a change in pull
request #821: [IOTDB-298]Last time-value query
Posted by GitBox <gi...@apache.org>.
JackieTien97 commented on a change in pull request #821: [IOTDB-298]Last time-value query
URL: https://github.com/apache/incubator-iotdb/pull/821#discussion_r380518625
##########
File path: server/src/main/java/org/apache/iotdb/db/metadata/MNode.java
##########
@@ -22,10 +22,15 @@
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.Map;
+
+import com.sun.rowset.internal.Row;
Review comment:
what's this for?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] wshao08 commented on a change in pull request
#821: [IOTDB-298]Last time-value query
Posted by GitBox <gi...@apache.org>.
wshao08 commented on a change in pull request #821: [IOTDB-298]Last time-value query
URL: https://github.com/apache/incubator-iotdb/pull/821#discussion_r386032620
##########
File path: server/src/main/java/org/apache/iotdb/db/metadata/mnode/MNode.java
##########
@@ -43,6 +44,8 @@
*/
protected String fullPath;
+ private TimeValuePair cachedLastValuePair = null;
Review comment:
Fixed
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] wshao08 commented on a change in pull request
#821: [IOTDB-298]Last time-value query
Posted by GitBox <gi...@apache.org>.
wshao08 commented on a change in pull request #821: [IOTDB-298]Last time-value query
URL: https://github.com/apache/incubator-iotdb/pull/821#discussion_r386032769
##########
File path: server/src/main/java/org/apache/iotdb/db/engine/StorageEngine.java
##########
@@ -235,7 +235,6 @@ public void insert(InsertPlan insertPlan)
throw new QueryProcessException(e);
}
}
-
Review comment:
Fixed
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] JackieTien97 commented on a change in pull
request #821: [IOTDB-298]Last time-value query
Posted by GitBox <gi...@apache.org>.
JackieTien97 commented on a change in pull request #821: [IOTDB-298]Last time-value query
URL: https://github.com/apache/incubator-iotdb/pull/821#discussion_r381034539
##########
File path: server/src/main/java/org/apache/iotdb/db/qp/physical/crud/BatchInsertPlan.java
##########
@@ -288,6 +291,19 @@ public long getMaxTime() {
return maxTime;
}
+ public TimeValuePair composeLastTimeValuePair(int measurementIndex) {
+ long maxTime = Long.MIN_VALUE;
+ int maxIndex = 0;
+ for (int i = 0; i < times.length; i++) {
+ if (times[i] > maxTime) {
+ maxTime = times[i];
+ maxIndex = i;
+ }
+ }
Review comment:
No matter what measurementIndex is, this calculation process is the same. It only needs to be called once.
You can extract that method outside the the calling loop.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] qiaojialin commented on a change in pull request
#821: [IOTDB-298]Last time-value query
Posted by GitBox <gi...@apache.org>.
qiaojialin commented on a change in pull request #821: [IOTDB-298]Last time-value query
URL: https://github.com/apache/incubator-iotdb/pull/821#discussion_r384934680
##########
File path: server/src/main/java/org/apache/iotdb/db/service/TSServiceImpl.java
##########
@@ -916,6 +913,15 @@ else if (constMeasurementsLoc < plan.getConstMeasurements().size()
plan.setDataTypeConsistencyChecker(null);
}
+ private void getLastQueryHeaders(
Review comment:
I don't remember, doesn't the respColumns contain time column?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] wshao08 commented on a change in pull request
#821: [IOTDB-298]Last time-value query
Posted by GitBox <gi...@apache.org>.
wshao08 commented on a change in pull request #821: [IOTDB-298]Last time-value query
URL: https://github.com/apache/incubator-iotdb/pull/821#discussion_r385593426
##########
File path: server/src/main/java/org/apache/iotdb/db/metadata/mnode/MNode.java
##########
@@ -43,6 +44,8 @@
*/
protected String fullPath;
+ private TimeValuePair cachedLastValuePair = null;
Review comment:
Fixed
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] qiaojialin commented on a change in pull request
#821: [IOTDB-298]Last time-value query
Posted by GitBox <gi...@apache.org>.
qiaojialin commented on a change in pull request #821: [IOTDB-298]Last time-value query
URL: https://github.com/apache/incubator-iotdb/pull/821#discussion_r385675556
##########
File path: server/src/main/java/org/apache/iotdb/db/metadata/mnode/InternalMNode.java
##########
@@ -20,6 +20,8 @@
import java.util.LinkedHashMap;
import java.util.Map;
+
+import org.apache.iotdb.tsfile.read.TimeValuePair;
Review comment:
remove
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] wshao08 commented on a change in pull request
#821: [IOTDB-298]Last time-value query
Posted by GitBox <gi...@apache.org>.
wshao08 commented on a change in pull request #821: [IOTDB-298]Last time-value query
URL: https://github.com/apache/incubator-iotdb/pull/821#discussion_r386394817
##########
File path: server/src/main/java/org/apache/iotdb/db/utils/FileLoaderUtils.java
##########
@@ -64,4 +72,40 @@ public static void updateTsFileResource(TsFileMetaData metaData, TsFileSequenceR
}
}
}
+
+ 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) {
+ if (data.getChunkLoader() == null) {
+ TsFileSequenceReader tsFileSequenceReader =
+ FileReaderManager.getInstance().get(resource, resource.isClosed());
+ data.setChunkLoader(new DiskChunkLoader(tsFileSequenceReader));
+ }
Review comment:
removed
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] qiaojialin commented on a change in pull request
#821: [IOTDB-298]Last time-value query
Posted by GitBox <gi...@apache.org>.
qiaojialin commented on a change in pull request #821: [IOTDB-298]Last time-value query
URL: https://github.com/apache/incubator-iotdb/pull/821#discussion_r385675814
##########
File path: server/src/main/java/org/apache/iotdb/db/metadata/mnode/LeafMNode.java
##########
@@ -74,4 +77,25 @@ public int getLeafCount() {
public MeasurementSchema getSchema() {
return schema;
}
+
+ public TimeValuePair getCachedLast() {
+ return cachedLastValuePair;
+ }
+
+ public synchronized void updateCachedLast(TimeValuePair timeValuePair, boolean insertionUpdate,
+ Long latestFlushedTime) {
+ if (timeValuePair == null || timeValuePair.getValue() == null) return;
+
+ if (cachedLastValuePair == null){
+ // If no cached last, a last read or an insertion to a sequenceTsFile will update cache.
+ if (!insertionUpdate || latestFlushedTime < timeValuePair.getTimestamp()) {
Review comment:
```suggestion
if (!insertionUpdate || latestFlushedTime <= timeValuePair.getTimestamp()) {
```
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] wshao08 closed pull request #821: [IOTDB-298]Last
time-value query
Posted by GitBox <gi...@apache.org>.
wshao08 closed pull request #821: [IOTDB-298]Last time-value query
URL: https://github.com/apache/incubator-iotdb/pull/821
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] wshao08 commented on a change in pull request
#821: [IOTDB-298]Last time-value query
Posted by GitBox <gi...@apache.org>.
wshao08 commented on a change in pull request #821: [IOTDB-298]Last time-value query
URL: https://github.com/apache/incubator-iotdb/pull/821#discussion_r380796485
##########
File path: server/src/main/java/org/apache/iotdb/db/qp/physical/crud/BatchInsertPlan.java
##########
@@ -288,6 +292,23 @@ public long getMaxTime() {
return maxTime;
}
+ public void updateMNodeLastValues(MNode node) {
+ long maxTime = Long.MIN_VALUE;
+ int maxIndex = 0;
+ for (int i = 0; i < times.length; i++) {
+ if (times[i] > maxTime) {
+ maxTime = times[i];
+ maxIndex = i;
+ }
+ }
+ for (int i = 0; i < measurements.length; i++) {
+ if (node.hasChild(measurements[i])) {
+ Object[] column = (Object[]) columns[i];
+ node.getChild(measurements[i]).updateCachedLast(maxTime, column[maxIndex], dataTypes[i]);
+ }
+ }
+ }
+
Review comment:
Agree. Moved it out of InsertPlan
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] qiaojialin commented on a change in pull request
#821: [IOTDB-298]Last time-value query
Posted by GitBox <gi...@apache.org>.
qiaojialin commented on a change in pull request #821: [IOTDB-298]Last time-value query
URL: https://github.com/apache/incubator-iotdb/pull/821#discussion_r385681751
##########
File path: server/src/main/java/org/apache/iotdb/db/query/executor/LastQueryExecutor.java
##########
@@ -0,0 +1,202 @@
+/*
+ * 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.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);
+
+ for (int i = seqFileResources.size() - 1; i >= 0; i--) {
+ List<ChunkMetaData> chunkMetadata =
+ loadSatisfiedChunkMetadata(seqFileResources.get(i), seriesPath, context);
+ if (!chunkMetadata.isEmpty()) {
+ ChunkMetaData lastChunkMetaData = chunkMetadata.get(chunkMetadata.size() - 1);
+ Statistics chunkStatistics = lastChunkMetaData.getStatistics();
+ resultPair = constructLastPair(
+ chunkStatistics.getEndTime(), chunkStatistics.getLastValue(), tsDataType);
+ break;
+ }
+ }
+
+ long version = 0;
+ for (TsFileResource resource : unseqFileResources) {
+ if (resource.getEndTimeMap().get(seriesPath.getDevice()) < resultPair.getTimestamp()) {
+ break;
+ }
+ List<ChunkMetaData> chunkMetadata = loadSatisfiedChunkMetadata(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 List<ChunkMetaData> loadSatisfiedChunkMetadata(
+ 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) {
+ if (data.getChunkLoader() == null) {
+ 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;
+ }
+
+ private TimeValuePair constructLastPair(long timestamp, Object value, TSDataType dataType) {
+ TimeValuePair lastPair = new TimeValuePair(timestamp, TsPrimitiveType.getByType(dataType, value));
+ return lastPair;
Review comment:
```suggestion
return TimeValuePair(timestamp, TsPrimitiveType.getByType(dataType, value));
```
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] qiaojialin commented on a change in pull request
#821: [IOTDB-298]Last time-value query
Posted by GitBox <gi...@apache.org>.
qiaojialin commented on a change in pull request #821: [IOTDB-298]Last time-value query
URL: https://github.com/apache/incubator-iotdb/pull/821#discussion_r384931514
##########
File path: server/src/main/java/org/apache/iotdb/db/qp/strategy/PhysicalGenerator.java
##########
@@ -236,6 +237,8 @@ private PhysicalPlan transformQuery(QueryOperator queryOperator)
queryPlan = new AggregationPlan();
((AggregationPlan) queryPlan)
.setAggregations(queryOperator.getSelectOperator().getAggregations());
+ } else if (queryOperator.hasLast()) {
Review comment:
```suggestion
} else if (queryOperator.isLastQuery()) {
```
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] qiaojialin commented on a change in pull request
#821: [IOTDB-298]Last time-value query
Posted by GitBox <gi...@apache.org>.
qiaojialin commented on a change in pull request #821: [IOTDB-298]Last time-value query
URL: https://github.com/apache/incubator-iotdb/pull/821#discussion_r386331477
##########
File path: 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缓存更新的逻辑位于`LeafNode`的`updateCachedLast`方法内，这里引入两个额外的参数`highPriorityUpdate`和`latestFlushTime`。`highPriorityUpdate`用来表示本次更新是否是高优先级的，新数据写入而导致的缓存更新都被认为是高优先级更新，而查询时更新缓存默认为低优先级更新。`latestFlushTime`用来记录当前已被写回到磁盘的数据的最大时间戳。
Review comment:
```suggestion
Last缓存更新的逻辑位于`LeafMNode`的`updateCachedLast`方法内，这里引入两个额外的参数`highPriorityUpdate`和`latestFlushTime`。`highPriorityUpdate`用来表示本次更新是否是高优先级的，新数据写入而导致的缓存更新都被认为是高优先级更新，而查询时更新缓存默认为低优先级更新。`latestFlushTime`用来记录当前已被写回到磁盘的数据的最大时间戳。
```
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] qiaojialin commented on a change in pull request
#821: [IOTDB-298]Last time-value query
Posted by GitBox <gi...@apache.org>.
qiaojialin commented on a change in pull request #821: [IOTDB-298]Last time-value query
URL: https://github.com/apache/incubator-iotdb/pull/821#discussion_r384930761
##########
File path: server/src/main/java/org/apache/iotdb/db/metadata/mnode/MNode.java
##########
@@ -113,4 +116,22 @@ public String getName() {
public void setName(String name) {
this.name = name;
}
-}
\ No newline at end of file
+
+ public TimeValuePair getCachedLast() {
+ return cachedLastValuePair;
+ }
+
+ public synchronized void updateCachedLast(
Review comment:
add javadoc for highPriorityUpdate
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] JackieTien97 commented on a change in pull
request #821: [IOTDB-298]Last time-value query
Posted by GitBox <gi...@apache.org>.
JackieTien97 commented on a change in pull request #821: [IOTDB-298]Last time-value query
URL: https://github.com/apache/incubator-iotdb/pull/821#discussion_r381032153
##########
File path: server/src/main/java/org/apache/iotdb/db/qp/executor/PlanExecutor.java
##########
@@ -772,8 +779,12 @@ private void checkPathExists(MNode node, String fullPath, MeasurementSchema sche
measurementNode.getSchema().getType()));
}
}
- return storageEngine.insertBatch(batchInsertPlan);
-
+ Integer[] results = storageEngine.insertBatch(batchInsertPlan);
+ for (int i = 0; i < measurementList.length; i++) {
+ MNode measurementNode = node.getChild(measurementList[i]);
+ measurementNode.updateCachedLast(batchInsertPlan.composeLastTimeValuePair(i));
+ }
Review comment:
same as before.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] wshao08 commented on a change in pull request
#821: [IOTDB-298]Last time-value query
Posted by GitBox <gi...@apache.org>.
wshao08 commented on a change in pull request #821: [IOTDB-298]Last time-value query
URL: https://github.com/apache/incubator-iotdb/pull/821#discussion_r380815684
##########
File path: server/src/main/java/org/apache/iotdb/db/query/executor/LastQueryExecutor.java
##########
@@ -0,0 +1,201 @@
+/*
+ * 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 org.apache.iotdb.db.exception.StorageEngineException;
+import org.apache.iotdb.db.exception.path.PathException;
+import org.apache.iotdb.db.exception.query.QueryProcessException;
+import org.apache.iotdb.db.metadata.MManager;
+import org.apache.iotdb.db.metadata.MNode;
+import org.apache.iotdb.db.qp.constant.SQLConstant;
+import org.apache.iotdb.db.qp.physical.crud.AggregationPlan;
+import org.apache.iotdb.db.qp.physical.crud.LastQueryPlan;
+import org.apache.iotdb.db.query.aggregation.AggregateResult;
+import org.apache.iotdb.db.query.context.QueryContext;
+import org.apache.iotdb.db.query.control.QueryResourceManager;
+import org.apache.iotdb.db.query.dataset.SingleDataSet;
+import org.apache.iotdb.db.query.factory.AggreResultFactory;
+import org.apache.iotdb.db.query.reader.series.IAggregateReader;
+import org.apache.iotdb.db.query.reader.series.SeriesAggregateReader;
+import org.apache.iotdb.tsfile.exception.cache.CacheException;
+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.common.BatchData;
+import org.apache.iotdb.tsfile.read.common.Path;
+import org.apache.iotdb.tsfile.read.common.RowRecord;
+import org.apache.iotdb.tsfile.read.filter.basic.Filter;
+import org.apache.iotdb.tsfile.read.query.dataset.QueryDataSet;
+import org.apache.iotdb.tsfile.utils.TsPrimitiveType;
+import org.omg.CORBA.TIMEOUT;
+
+import java.io.IOException;
+import java.sql.Time;
+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 {
+
+ List<LastQueryResult> lastQueryResultList = new ArrayList<>();
+ for (int i = 0; i < selectedSeries.size(); i++) {
+ LastQueryResult lastQueryResult = calculateLastPairForOneSeries(selectedSeries.get(i), dataTypes.get(i), context);
+ lastQueryResultList.add(lastQueryResult);
+ }
+
+ RowRecord resultRecord = constructLastRowRecord(lastQueryResultList);
+ SingleDataSet dataSet = new SingleDataSet(selectedSeries, dataTypes);
+ dataSet.setRecord(resultRecord);
+ return dataSet;
+ }
+
+ /**
+ * get aggregation result for one series
+ *
+ * @param context query context
+ * @return AggregateResult list
+ */
+ private LastQueryResult calculateLastPairForOneSeries(
+ Path seriesPath, TSDataType tsDataType,
+ QueryContext context)
+ throws IOException, QueryProcessException, StorageEngineException {
+ LastQueryResult queryResult = new LastQueryResult();
+ MNode node = null;
+ try {
+ node = MManager.getInstance().getNodeByPathFromCache(seriesPath.toString());
+ } catch (PathException e) {
+ throw new QueryProcessException(e);
+ } catch (CacheException e) {
+ throw new QueryProcessException(e.getMessage());
+ }
+ if (node.getCachedLast() != null) {
+ queryResult.setPairResult(node.getCachedLast());
+ return queryResult;
+ }
+
+ // construct series reader without value filter
+ Filter timeFilter = null;
+ IAggregateReader seriesReader = new SeriesAggregateReader(
+ seriesPath, tsDataType, context, QueryResourceManager.getInstance()
+ .getQueryDataSource(seriesPath, context, timeFilter), timeFilter, null);
+
+ long maxTime = Long.MIN_VALUE;
+ while (seriesReader.hasNextChunk()) {
+ while (seriesReader.hasNextPage()) {
+ // cal by page data
+ while (seriesReader.hasNextOverlappedPage()) {
+ BatchData nextOverlappedPageData = seriesReader.nextOverlappedPage();
+ int maxIndex = nextOverlappedPageData.length() - 1;
+ if (maxIndex < 0) {
+ continue;
+ }
+ long time = nextOverlappedPageData.getTimeByIndex(maxIndex);
+ if (time > maxTime) {
+ maxTime = time;
+ queryResult.setPairResult(maxTime, nextOverlappedPageData.getValueInTimestamp(time), tsDataType);
+ }
+ nextOverlappedPageData.resetBatchData();
+ }
+ }
+ }
Review comment:
Yes, adopted
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] qiaojialin commented on a change in pull request
#821: [IOTDB-298]Last time-value query
Posted by GitBox <gi...@apache.org>.
qiaojialin commented on a change in pull request #821: [IOTDB-298]Last time-value query
URL: https://github.com/apache/incubator-iotdb/pull/821#discussion_r385673726
##########
File path: server/src/main/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessor.java
##########
@@ -569,18 +607,40 @@ public void insert(InsertPlan insertPlan) throws QueryProcessException {
}
}
+ long latestFlushedTime = globalLatestFlushedTimeForEachDevice.computeIfAbsent(
+ batchInsertPlan.getDeviceId(), k -> Long.MIN_VALUE);
// do not forget last part
if (before < loc) {
insertBatchToTsFileProcessor(batchInsertPlan, before, loc, isSequence, results,
beforeTimePartition);
}
+ updateBatchInsertPlanLast(batchInsertPlan, latestFlushedTime);
+ if (latestFlushedTime < batchInsertPlan.getMaxTime())
+ globalLatestFlushedTimeForEachDevice.put(batchInsertPlan.getDeviceId(), batchInsertPlan.getMaxTime());
Review comment:
move this into insertBatchToTsFileProcessor()
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] wshao08 commented on a change in pull request
#821: [IOTDB-298]Last time-value query
Posted by GitBox <gi...@apache.org>.
wshao08 commented on a change in pull request #821: [IOTDB-298]Last time-value query
URL: https://github.com/apache/incubator-iotdb/pull/821#discussion_r384646457
##########
File path: server/src/main/java/org/apache/iotdb/db/query/executor/LastQueryExecutor.java
##########
@@ -0,0 +1,216 @@
+/*
+ * 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 org.apache.iotdb.db.exception.StorageEngineException;
+import org.apache.iotdb.db.exception.path.PathException;
+import org.apache.iotdb.db.exception.query.QueryProcessException;
+import org.apache.iotdb.db.metadata.MManager;
+import org.apache.iotdb.db.metadata.MNode;
+import org.apache.iotdb.db.qp.physical.crud.LastQueryPlan;
+import org.apache.iotdb.db.query.aggregation.AggregateResult;
+import org.apache.iotdb.db.query.context.QueryContext;
+import org.apache.iotdb.db.query.control.QueryResourceManager;
+import org.apache.iotdb.db.query.dataset.SingleDataSet;
+import org.apache.iotdb.db.query.reader.series.IAggregateReader;
+import org.apache.iotdb.db.query.reader.series.SeriesAggregateReader;
+import org.apache.iotdb.tsfile.exception.cache.CacheException;
+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.common.BatchData;
+import org.apache.iotdb.tsfile.read.common.Path;
+import org.apache.iotdb.tsfile.read.common.RowRecord;
+import org.apache.iotdb.tsfile.read.filter.basic.Filter;
+import org.apache.iotdb.tsfile.read.query.dataset.QueryDataSet;
+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 {
+
+ List<LastQueryResult> lastQueryResultList = new ArrayList<>();
+ for (int i = 0; i < selectedSeries.size(); i++) {
+ LastQueryResult lastQueryResult = calculateLastPairForOneSeries(selectedSeries.get(i), dataTypes.get(i), context);
+ lastQueryResultList.add(lastQueryResult);
+ }
+
+ RowRecord resultRecord = constructLastRowRecord(lastQueryResultList);
+ SingleDataSet dataSet = new SingleDataSet(selectedSeries, dataTypes);
+ dataSet.setRecord(resultRecord);
+ return dataSet;
+ }
+
+ /**
+ * get aggregation result for one series
+ *
+ * @param context query context
+ * @return AggregateResult list
+ */
+ private LastQueryResult calculateLastPairForOneSeries(
+ Path seriesPath, TSDataType tsDataType,
+ QueryContext context)
+ throws IOException, QueryProcessException, StorageEngineException {
+ LastQueryResult queryResult = new LastQueryResult();
+ MNode node = null;
+ try {
+ node = MManager.getInstance().getNodeByPathFromCache(seriesPath.toString());
+ } catch (PathException e) {
+ throw new QueryProcessException(e);
+ } catch (CacheException e) {
+ throw new QueryProcessException(e.getMessage());
+ }
+ if (node.getCachedLast() != null) {
+ queryResult.setPairResult(node.getCachedLast());
+ return queryResult;
+ }
+
+ // construct series reader without value filter
+ Filter timeFilter = null;
+ IAggregateReader seriesReader = new SeriesAggregateReader(
+ seriesPath, tsDataType, context, QueryResourceManager.getInstance()
+ .getQueryDataSource(seriesPath, context, timeFilter), timeFilter, null);
+
+ long maxTime = Long.MIN_VALUE;
+ while (seriesReader.hasNextChunk()) {
+ // cal by chunk statistics
+ if (seriesReader.canUseCurrentChunkStatistics()) {
+ Statistics chunkStatistics = seriesReader.currentChunkStatistics();
+ if (chunkStatistics.getEndTime() > maxTime) {
+ maxTime = chunkStatistics.getEndTime();
+ queryResult.setPairResult(maxTime, chunkStatistics.getLastValue(), tsDataType);
+ }
+ seriesReader.skipCurrentChunk();
+ continue;
+ }
+ while (seriesReader.hasNextPage()) {
+ //cal by page statistics
+ if (seriesReader.canUseCurrentPageStatistics()) {
+ Statistics pageStatistic = seriesReader.currentPageStatistics();
+ if (pageStatistic.getEndTime() > maxTime) {
+ maxTime = pageStatistic.getEndTime();
+ queryResult.setPairResult(maxTime, pageStatistic.getLastValue(), tsDataType);
+ }
+ seriesReader.skipCurrentPage();
+ continue;
+ }
+ // cal by page data
+ while (seriesReader.hasNextOverlappedPage()) {
+ BatchData nextOverlappedPageData = seriesReader.nextOverlappedPage();
+ int maxIndex = nextOverlappedPageData.length() - 1;
+ if (maxIndex < 0) {
+ continue;
+ }
+ long time = nextOverlappedPageData.getTimeByIndex(maxIndex);
+ if (time > maxTime) {
+ maxTime = time;
+ queryResult.setPairResult(maxTime, nextOverlappedPageData.getValueInTimestamp(time), tsDataType);
+ }
+ nextOverlappedPageData.resetBatchData();
+ }
+ }
+ }
+ if (queryResult.hasResult())
+ node.setCachedLast(queryResult.getPairResult());
Review comment:
Updated, thanks
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] qiaojialin commented on a change in pull request
#821: [IOTDB-298]Last time-value query
Posted by GitBox <gi...@apache.org>.
qiaojialin commented on a change in pull request #821: [IOTDB-298]Last time-value query
URL: https://github.com/apache/incubator-iotdb/pull/821#discussion_r385682407
##########
File path: server/src/main/java/org/apache/iotdb/db/query/executor/LastQueryExecutor.java
##########
@@ -0,0 +1,202 @@
+/*
+ * 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.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);
+
+ for (int i = seqFileResources.size() - 1; i >= 0; i--) {
+ List<ChunkMetaData> chunkMetadata =
+ loadSatisfiedChunkMetadata(seqFileResources.get(i), seriesPath, context);
+ if (!chunkMetadata.isEmpty()) {
+ ChunkMetaData lastChunkMetaData = chunkMetadata.get(chunkMetadata.size() - 1);
+ Statistics chunkStatistics = lastChunkMetaData.getStatistics();
+ resultPair = constructLastPair(
+ chunkStatistics.getEndTime(), chunkStatistics.getLastValue(), tsDataType);
+ break;
+ }
+ }
+
+ long version = 0;
+ for (TsFileResource resource : unseqFileResources) {
+ if (resource.getEndTimeMap().get(seriesPath.getDevice()) < resultPair.getTimestamp()) {
+ break;
+ }
+ List<ChunkMetaData> chunkMetadata = loadSatisfiedChunkMetadata(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);
Review comment:
when I saw this line, I think maybe move the lastFlushTime out of the updateCachedLast mehod and check in the insertion is better
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] wshao08 commented on a change in pull request
#821: [IOTDB-298]Last time-value query
Posted by GitBox <gi...@apache.org>.
wshao08 commented on a change in pull request #821: [IOTDB-298]Last time-value query
URL: https://github.com/apache/incubator-iotdb/pull/821#discussion_r386032819
##########
File path: server/src/main/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessor.java
##########
@@ -496,15 +511,38 @@ public void insert(InsertPlan insertPlan) throws QueryProcessException {
latestFlushedTimeForEachDevice.computeIfAbsent(timePartitionId, id -> new HashMap<>())
.putIfAbsent(insertPlan.getDeviceId(), Long.MIN_VALUE);
+ long latestFlushedTime = globalLatestFlushedTimeForEachDevice.computeIfAbsent(
+ insertPlan.getDeviceId(), k -> Long.MIN_VALUE);
// insert to sequence or unSequence file
insertToTsFileProcessor(insertPlan,
insertPlan.getTime() > latestFlushedTimeForEachDevice.get(timePartitionId)
.get(insertPlan.getDeviceId()));
+
+ updateInsertPlanLast(insertPlan, latestFlushedTime);
+ if (latestFlushedTime < insertPlan.getTime())
+ globalLatestFlushedTimeForEachDevice.put(insertPlan.getDeviceId(), insertPlan.getTime());
Review comment:
Yes moved, looks better now.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] JackieTien97 commented on a change in pull
request #821: [IOTDB-298]Last time-value query
Posted by GitBox <gi...@apache.org>.
JackieTien97 commented on a change in pull request #821: [IOTDB-298]Last time-value query
URL: https://github.com/apache/incubator-iotdb/pull/821#discussion_r380539293
##########
File path: server/src/main/java/org/apache/iotdb/db/qp/logical/Operator.java
##########
@@ -68,7 +68,7 @@ public String toString() {
public enum OperatorType {
SFW, JOIN, UNION, FILTER, GROUPBY, ORDERBY, LIMIT, SELECT, SEQTABLESCAN, HASHTABLESCAN,
MERGEJOIN, FILEREAD, NULL, TABLESCAN, UPDATE, INSERT, BATCHINSERT, DELETE, BASIC_FUNC, IN, QUERY, MERGEQUERY,
- AGGREGATION, AUTHOR, FROM, FUNC, LOADDATA, METADATA, PROPERTY, INDEX, INDEXQUERY, FILL,
+ AGGREGATION, AUTHOR, FROM, FUNC, LOADDATA, METADATA, PROPERTY, INDEX, INDEXQUERY, FILL, LAST,
Review comment:
It's better to add it to the last, because you don't know whether the programmers are using the enum's ordinal() function
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] wshao08 commented on a change in pull request
#821: [IOTDB-298]Last time-value query
Posted by GitBox <gi...@apache.org>.
wshao08 commented on a change in pull request #821: [IOTDB-298]Last time-value query
URL: https://github.com/apache/incubator-iotdb/pull/821#discussion_r380796650
##########
File path: server/src/main/java/org/apache/iotdb/db/qp/physical/crud/InsertPlan.java
##########
@@ -185,4 +190,13 @@ public void deserializeFrom(ByteBuffer buffer) {
public String toString() {
return "deviceId: " + deviceId + ", time: " + time;
}
+
+ public void updateMNodeLastValues(MNode node) throws QueryProcessException {
+ for (int i = 0; i < measurements.length; i++) {
+ if (node.hasChild(measurements[i])) {
+ Object value = CommonUtils.parseValue(dataTypes[i], values[i]);
+ node.getChild(measurements[i]).updateCachedLast(time, value, dataTypes[i]);
+ }
+ }
+ }
Review comment:
Fixed
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] qiaojialin commented on a change in pull request
#821: [IOTDB-298]Last time-value query
Posted by GitBox <gi...@apache.org>.
qiaojialin commented on a change in pull request #821: [IOTDB-298]Last time-value query
URL: https://github.com/apache/incubator-iotdb/pull/821#discussion_r382467091
##########
File path: server/src/main/java/org/apache/iotdb/db/query/executor/LastQueryExecutor.java
##########
@@ -0,0 +1,216 @@
+/*
+ * 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 org.apache.iotdb.db.exception.StorageEngineException;
+import org.apache.iotdb.db.exception.path.PathException;
+import org.apache.iotdb.db.exception.query.QueryProcessException;
+import org.apache.iotdb.db.metadata.MManager;
+import org.apache.iotdb.db.metadata.MNode;
+import org.apache.iotdb.db.qp.physical.crud.LastQueryPlan;
+import org.apache.iotdb.db.query.aggregation.AggregateResult;
+import org.apache.iotdb.db.query.context.QueryContext;
+import org.apache.iotdb.db.query.control.QueryResourceManager;
+import org.apache.iotdb.db.query.dataset.SingleDataSet;
+import org.apache.iotdb.db.query.reader.series.IAggregateReader;
+import org.apache.iotdb.db.query.reader.series.SeriesAggregateReader;
+import org.apache.iotdb.tsfile.exception.cache.CacheException;
+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.common.BatchData;
+import org.apache.iotdb.tsfile.read.common.Path;
+import org.apache.iotdb.tsfile.read.common.RowRecord;
+import org.apache.iotdb.tsfile.read.filter.basic.Filter;
+import org.apache.iotdb.tsfile.read.query.dataset.QueryDataSet;
+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 {
+
+ List<LastQueryResult> lastQueryResultList = new ArrayList<>();
+ for (int i = 0; i < selectedSeries.size(); i++) {
+ LastQueryResult lastQueryResult = calculateLastPairForOneSeries(selectedSeries.get(i), dataTypes.get(i), context);
+ lastQueryResultList.add(lastQueryResult);
+ }
+
+ RowRecord resultRecord = constructLastRowRecord(lastQueryResultList);
+ SingleDataSet dataSet = new SingleDataSet(selectedSeries, dataTypes);
+ dataSet.setRecord(resultRecord);
+ return dataSet;
+ }
+
+ /**
+ * get aggregation result for one series
+ *
+ * @param context query context
+ * @return AggregateResult list
+ */
+ private LastQueryResult calculateLastPairForOneSeries(
+ Path seriesPath, TSDataType tsDataType,
+ QueryContext context)
+ throws IOException, QueryProcessException, StorageEngineException {
+ LastQueryResult queryResult = new LastQueryResult();
+ MNode node = null;
+ try {
+ node = MManager.getInstance().getNodeByPathFromCache(seriesPath.toString());
+ } catch (PathException e) {
+ throw new QueryProcessException(e);
+ } catch (CacheException e) {
+ throw new QueryProcessException(e.getMessage());
+ }
+ if (node.getCachedLast() != null) {
+ queryResult.setPairResult(node.getCachedLast());
+ return queryResult;
+ }
+
+ // construct series reader without value filter
+ Filter timeFilter = null;
+ IAggregateReader seriesReader = new SeriesAggregateReader(
+ seriesPath, tsDataType, context, QueryResourceManager.getInstance()
+ .getQueryDataSource(seriesPath, context, timeFilter), timeFilter, null);
+
+ long maxTime = Long.MIN_VALUE;
+ while (seriesReader.hasNextChunk()) {
+ // cal by chunk statistics
+ if (seriesReader.canUseCurrentChunkStatistics()) {
+ Statistics chunkStatistics = seriesReader.currentChunkStatistics();
+ if (chunkStatistics.getEndTime() > maxTime) {
+ maxTime = chunkStatistics.getEndTime();
+ queryResult.setPairResult(maxTime, chunkStatistics.getLastValue(), tsDataType);
+ }
+ seriesReader.skipCurrentChunk();
+ continue;
+ }
+ while (seriesReader.hasNextPage()) {
+ //cal by page statistics
+ if (seriesReader.canUseCurrentPageStatistics()) {
+ Statistics pageStatistic = seriesReader.currentPageStatistics();
+ if (pageStatistic.getEndTime() > maxTime) {
+ maxTime = pageStatistic.getEndTime();
+ queryResult.setPairResult(maxTime, pageStatistic.getLastValue(), tsDataType);
+ }
+ seriesReader.skipCurrentPage();
+ continue;
+ }
+ // cal by page data
+ while (seriesReader.hasNextOverlappedPage()) {
+ BatchData nextOverlappedPageData = seriesReader.nextOverlappedPage();
+ int maxIndex = nextOverlappedPageData.length() - 1;
+ if (maxIndex < 0) {
+ continue;
+ }
+ long time = nextOverlappedPageData.getTimeByIndex(maxIndex);
+ if (time > maxTime) {
+ maxTime = time;
+ queryResult.setPairResult(maxTime, nextOverlappedPageData.getValueInTimestamp(time), tsDataType);
+ }
+ nextOverlappedPageData.resetBatchData();
+ }
+ }
+ }
+ if (queryResult.hasResult())
+ node.setCachedLast(queryResult.getPairResult());
+ return queryResult;
+ }
+
+ /**
+ * using last result data list construct QueryDataSet.
+ *
+ * @param lastQueryResultList last result list
+ */
+ private RowRecord constructLastRowRecord(List<LastQueryResult> lastQueryResultList) {
+ long maxTime = Long.MIN_VALUE;
+ for (LastQueryResult lastPair : lastQueryResultList) {
+ if (lastPair.hasResult() && lastPair.getTimestamp() > maxTime)
+ maxTime = lastPair.getTimestamp();
+ }
+
+ RowRecord resultRecord = new RowRecord(maxTime);
+ for (int i = 0; i < lastQueryResultList.size(); i++) {
+ TSDataType dataType = dataTypes.get(i);
+ LastQueryResult lastPair = lastQueryResultList.get(i);
+ if (lastPair.hasResult() && lastPair.getTimestamp() == maxTime)
+ resultRecord.addField(lastPair.getValue(), dataType);
+ else
+ resultRecord.addField(null, dataType);
+ }
+
+ return resultRecord;
+ }
+
+ class LastQueryResult {
Review comment:
I think this class is not needed, you could use TimeValuePair directly unless you store the path of a series in this class.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] wshao08 commented on a change in pull request
#821: [IOTDB-298]Last time-value query
Posted by GitBox <gi...@apache.org>.
wshao08 commented on a change in pull request #821: [IOTDB-298]Last time-value query
URL: https://github.com/apache/incubator-iotdb/pull/821#discussion_r385593560
##########
File path: server/src/main/java/org/apache/iotdb/db/query/executor/LastQueryExecutor.java
##########
@@ -0,0 +1,172 @@
+/*
+ * 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.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.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.QueryResourceManager;
+import org.apache.iotdb.db.query.dataset.ListDataSet;
+import org.apache.iotdb.db.query.reader.series.IAggregateReader;
+import org.apache.iotdb.db.query.reader.series.SeriesAggregateReader;
+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.common.BatchData;
+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(
Review comment:
Fixed
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] qiaojialin commented on a change in pull request
#821: [IOTDB-298]Last time-value query
Posted by GitBox <gi...@apache.org>.
qiaojialin commented on a change in pull request #821: [IOTDB-298]Last time-value query
URL: https://github.com/apache/incubator-iotdb/pull/821#discussion_r385679836
##########
File path: server/src/main/java/org/apache/iotdb/db/qp/logical/crud/SelectOperator.java
##########
@@ -50,6 +52,11 @@ public void addClusterPath(Path suffixPath, String aggregation) {
aggregations.add(aggregation);
}
+ public void addLastPath(Path suffixPath) {
Review comment:
I suggest use addSelectPath to replace this method and add a setLastQuery()
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] wshao08 commented on a change in pull request
#821: [IOTDB-298]Last time-value query
Posted by GitBox <gi...@apache.org>.
wshao08 commented on a change in pull request #821: [IOTDB-298]Last time-value query
URL: https://github.com/apache/incubator-iotdb/pull/821#discussion_r380596490
##########
File path: server/src/main/java/org/apache/iotdb/db/service/TSServiceImpl.java
##########
@@ -196,6 +196,7 @@ public static TSDataType getSeriesType(String path) throws QueryProcessException
case SQLConstant.FIRST_VALUE:
case SQLConstant.MIN_VALUE:
case SQLConstant.MAX_VALUE:
+ //case SQLConstant.LAST:
Review comment:
Fixed
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] wshao08 commented on a change in pull request
#821: [IOTDB-298]Last time-value query
Posted by GitBox <gi...@apache.org>.
wshao08 commented on a change in pull request #821: [IOTDB-298]Last time-value query
URL: https://github.com/apache/incubator-iotdb/pull/821#discussion_r386032639
##########
File path: server/src/main/java/org/apache/iotdb/db/metadata/mnode/LeafMNode.java
##########
@@ -74,4 +77,25 @@ public int getLeafCount() {
public MeasurementSchema getSchema() {
return schema;
}
+
+ public TimeValuePair getCachedLast() {
+ return cachedLastValuePair;
+ }
+
+ public synchronized void updateCachedLast(TimeValuePair timeValuePair, boolean insertionUpdate,
+ Long latestFlushedTime) {
+ if (timeValuePair == null || timeValuePair.getValue() == null) return;
+
+ if (cachedLastValuePair == null){
+ // If no cached last, a last read or an insertion to a sequenceTsFile will update cache.
+ if (!insertionUpdate || latestFlushedTime < timeValuePair.getTimestamp()) {
Review comment:
Fixed
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] wshao08 commented on a change in pull request
#821: [IOTDB-298]Last time-value query
Posted by GitBox <gi...@apache.org>.
wshao08 commented on a change in pull request #821: [IOTDB-298]Last time-value query
URL: https://github.com/apache/incubator-iotdb/pull/821#discussion_r384647020
##########
File path: server/src/main/java/org/apache/iotdb/db/query/executor/LastQueryExecutor.java
##########
@@ -0,0 +1,216 @@
+/*
+ * 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 org.apache.iotdb.db.exception.StorageEngineException;
+import org.apache.iotdb.db.exception.path.PathException;
+import org.apache.iotdb.db.exception.query.QueryProcessException;
+import org.apache.iotdb.db.metadata.MManager;
+import org.apache.iotdb.db.metadata.MNode;
+import org.apache.iotdb.db.qp.physical.crud.LastQueryPlan;
+import org.apache.iotdb.db.query.aggregation.AggregateResult;
+import org.apache.iotdb.db.query.context.QueryContext;
+import org.apache.iotdb.db.query.control.QueryResourceManager;
+import org.apache.iotdb.db.query.dataset.SingleDataSet;
+import org.apache.iotdb.db.query.reader.series.IAggregateReader;
+import org.apache.iotdb.db.query.reader.series.SeriesAggregateReader;
+import org.apache.iotdb.tsfile.exception.cache.CacheException;
+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.common.BatchData;
+import org.apache.iotdb.tsfile.read.common.Path;
+import org.apache.iotdb.tsfile.read.common.RowRecord;
+import org.apache.iotdb.tsfile.read.filter.basic.Filter;
+import org.apache.iotdb.tsfile.read.query.dataset.QueryDataSet;
+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 {
+
+ List<LastQueryResult> lastQueryResultList = new ArrayList<>();
+ for (int i = 0; i < selectedSeries.size(); i++) {
+ LastQueryResult lastQueryResult = calculateLastPairForOneSeries(selectedSeries.get(i), dataTypes.get(i), context);
+ lastQueryResultList.add(lastQueryResult);
+ }
+
+ RowRecord resultRecord = constructLastRowRecord(lastQueryResultList);
Review comment:
Result set updated
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] wshao08 commented on a change in pull request
#821: [IOTDB-298]Last time-value query
Posted by GitBox <gi...@apache.org>.
wshao08 commented on a change in pull request #821: [IOTDB-298]Last time-value query
URL: https://github.com/apache/incubator-iotdb/pull/821#discussion_r384646242
##########
File path: server/src/main/java/org/apache/iotdb/db/query/executor/LastQueryExecutor.java
##########
@@ -0,0 +1,216 @@
+/*
+ * 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 org.apache.iotdb.db.exception.StorageEngineException;
+import org.apache.iotdb.db.exception.path.PathException;
+import org.apache.iotdb.db.exception.query.QueryProcessException;
+import org.apache.iotdb.db.metadata.MManager;
+import org.apache.iotdb.db.metadata.MNode;
+import org.apache.iotdb.db.qp.physical.crud.LastQueryPlan;
+import org.apache.iotdb.db.query.aggregation.AggregateResult;
+import org.apache.iotdb.db.query.context.QueryContext;
+import org.apache.iotdb.db.query.control.QueryResourceManager;
+import org.apache.iotdb.db.query.dataset.SingleDataSet;
+import org.apache.iotdb.db.query.reader.series.IAggregateReader;
+import org.apache.iotdb.db.query.reader.series.SeriesAggregateReader;
+import org.apache.iotdb.tsfile.exception.cache.CacheException;
+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.common.BatchData;
+import org.apache.iotdb.tsfile.read.common.Path;
+import org.apache.iotdb.tsfile.read.common.RowRecord;
+import org.apache.iotdb.tsfile.read.filter.basic.Filter;
+import org.apache.iotdb.tsfile.read.query.dataset.QueryDataSet;
+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 {
+
+ List<LastQueryResult> lastQueryResultList = new ArrayList<>();
+ for (int i = 0; i < selectedSeries.size(); i++) {
+ LastQueryResult lastQueryResult = calculateLastPairForOneSeries(selectedSeries.get(i), dataTypes.get(i), context);
+ lastQueryResultList.add(lastQueryResult);
+ }
+
+ RowRecord resultRecord = constructLastRowRecord(lastQueryResultList);
+ SingleDataSet dataSet = new SingleDataSet(selectedSeries, dataTypes);
+ dataSet.setRecord(resultRecord);
+ return dataSet;
+ }
+
+ /**
+ * get aggregation result for one series
+ *
+ * @param context query context
+ * @return AggregateResult list
+ */
+ private LastQueryResult calculateLastPairForOneSeries(
+ Path seriesPath, TSDataType tsDataType,
+ QueryContext context)
+ throws IOException, QueryProcessException, StorageEngineException {
+ LastQueryResult queryResult = new LastQueryResult();
+ MNode node = null;
+ try {
+ node = MManager.getInstance().getNodeByPathFromCache(seriesPath.toString());
+ } catch (PathException e) {
+ throw new QueryProcessException(e);
+ } catch (CacheException e) {
+ throw new QueryProcessException(e.getMessage());
+ }
+ if (node.getCachedLast() != null) {
+ queryResult.setPairResult(node.getCachedLast());
+ return queryResult;
+ }
+
+ // construct series reader without value filter
+ Filter timeFilter = null;
Review comment:
Fixed
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] wshao08 commented on a change in pull request
#821: [IOTDB-298]Last time-value query
Posted by GitBox <gi...@apache.org>.
wshao08 commented on a change in pull request #821: [IOTDB-298]Last time-value query
URL: https://github.com/apache/incubator-iotdb/pull/821#discussion_r386032601
##########
File path: server/src/main/java/org/apache/iotdb/db/query/executor/LastQueryExecutor.java
##########
@@ -0,0 +1,202 @@
+/*
+ * 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.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);
+
+ for (int i = seqFileResources.size() - 1; i >= 0; i--) {
Review comment:
Fixed
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] wshao08 commented on a change in pull request
#821: [IOTDB-298]Last time-value query
Posted by GitBox <gi...@apache.org>.
wshao08 commented on a change in pull request #821: [IOTDB-298]Last time-value query
URL: https://github.com/apache/incubator-iotdb/pull/821#discussion_r380821805
##########
File path: server/src/main/java/org/apache/iotdb/db/qp/executor/PlanExecutor.java
##########
@@ -680,6 +674,7 @@ public void insert(InsertPlan insertPlan) throws QueryProcessException {
}
insertPlan.setDataTypes(dataTypes);
storageEngine.insert(insertPlan);
+ insertPlan.updateMNodeLastValues(node);
Review comment:
The update function is moved into MNode structure.
I added a loop to update the last value after storageEngine.insert(). It is possible that insert() could fail and throw exceptions. In such case the last value cache will not be updated.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] qiaojialin commented on a change in pull request
#821: [IOTDB-298]Last time-value query
Posted by GitBox <gi...@apache.org>.
qiaojialin commented on a change in pull request #821: [IOTDB-298]Last time-value query
URL: https://github.com/apache/incubator-iotdb/pull/821#discussion_r385679108
##########
File path: server/src/main/java/org/apache/iotdb/db/metadata/mnode/MNode.java
##########
@@ -43,6 +44,8 @@
*/
protected String fullPath;
+ private TimeValuePair cachedLastValuePair = null;
Review comment:
remove this and its import
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] JackieTien97 commented on a change in pull
request #821: [IOTDB-298]Last time-value query
Posted by GitBox <gi...@apache.org>.
JackieTien97 commented on a change in pull request #821: [IOTDB-298]Last time-value query
URL: https://github.com/apache/incubator-iotdb/pull/821#discussion_r380540330
##########
File path: server/src/main/java/org/apache/iotdb/db/qp/physical/crud/InsertPlan.java
##########
@@ -185,4 +190,13 @@ public void deserializeFrom(ByteBuffer buffer) {
public String toString() {
return "deviceId: " + deviceId + ", time: " + time;
}
+
+ public void updateMNodeLastValues(MNode node) throws QueryProcessException {
+ for (int i = 0; i < measurements.length; i++) {
+ if (node.hasChild(measurements[i])) {
+ Object value = CommonUtils.parseValue(dataTypes[i], values[i]);
+ node.getChild(measurements[i]).updateCachedLast(time, value, dataTypes[i]);
+ }
+ }
+ }
Review comment:
Same as before
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] qiaojialin commented on a change in pull request
#821: [IOTDB-298]Last time-value query
Posted by GitBox <gi...@apache.org>.
qiaojialin commented on a change in pull request #821: [IOTDB-298]Last time-value query
URL: https://github.com/apache/incubator-iotdb/pull/821#discussion_r382463732
##########
File path: server/src/main/java/org/apache/iotdb/db/qp/physical/crud/BatchInsertPlan.java
##########
@@ -288,6 +291,19 @@ public long getMaxTime() {
return maxTime;
}
+ public TimeValuePair composeLastTimeValuePair(int measurementIndex) {
+ long maxTime = Long.MIN_VALUE;
+ int maxIndex = 0;
+ for (int i = 0; i < times.length; i++) {
Review comment:
the time in BatchInsertPlan is always in ascending order, you could directly get the last
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] JackieTien97 commented on a change in pull
request #821: [IOTDB-298]Last time-value query
Posted by GitBox <gi...@apache.org>.
JackieTien97 commented on a change in pull request #821: [IOTDB-298]Last time-value query
URL: https://github.com/apache/incubator-iotdb/pull/821#discussion_r380537858
##########
File path: server/src/main/java/org/apache/iotdb/db/qp/executor/PlanExecutor.java
##########
@@ -680,6 +674,7 @@ public void insert(InsertPlan insertPlan) throws QueryProcessException {
}
insertPlan.setDataTypes(dataTypes);
storageEngine.insert(insertPlan);
+ insertPlan.updateMNodeLastValues(node);
Review comment:
It is not suitable to put the updateMNodeLastValues method in the InsertPlan.
You can update the last value in the before loop, there you already got the measurementNode.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] qiaojialin commented on a change in pull request
#821: [IOTDB-298]Last time-value query
Posted by GitBox <gi...@apache.org>.
qiaojialin commented on a change in pull request #821: [IOTDB-298]Last time-value query
URL: https://github.com/apache/incubator-iotdb/pull/821#discussion_r385666771
##########
File path: server/src/main/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessor.java
##########
@@ -496,15 +511,38 @@ public void insert(InsertPlan insertPlan) throws QueryProcessException {
latestFlushedTimeForEachDevice.computeIfAbsent(timePartitionId, id -> new HashMap<>())
.putIfAbsent(insertPlan.getDeviceId(), Long.MIN_VALUE);
+ long latestFlushedTime = globalLatestFlushedTimeForEachDevice.computeIfAbsent(
Review comment:
```suggestion
long globalLatestFlushTime = globalLatestFlushedTimeForEachDevice.computeIfAbsent(
```
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] qiaojialin commented on a change in pull request
#821: [IOTDB-298]Last time-value query
Posted by GitBox <gi...@apache.org>.
qiaojialin commented on a change in pull request #821: [IOTDB-298]Last time-value query
URL: https://github.com/apache/incubator-iotdb/pull/821#discussion_r385661822
##########
File path: server/src/main/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessor.java
##########
@@ -178,6 +180,12 @@
* file.
*/
private Map<Long, Map<String, Long>> latestFlushedTimeForEachDevice = new HashMap<>();
Review comment:
rename this to latestDeviceFlushedTimeInEachPartition distinguish with globalLatestFlushedTimeForEachDevice
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] wshao08 commented on a change in pull request
#821: [IOTDB-298]Last time-value query
Posted by GitBox <gi...@apache.org>.
wshao08 commented on a change in pull request #821: [IOTDB-298]Last time-value query
URL: https://github.com/apache/incubator-iotdb/pull/821#discussion_r384646838
##########
File path: server/src/main/java/org/apache/iotdb/db/qp/strategy/LogicalGenerator.java
##########
@@ -1047,6 +1048,16 @@ public void enterSelectElement(SelectElementContext ctx) {
queryOp.setSelectOperator(selectOp);
}
+ @Override
+ public void enterLastElement(SqlBaseParser.LastElementContext ctx) {
+ super.enterLastElement(ctx);
+ selectOp = new SelectOperator(SQLConstant.TOK_SELECT);
+ LastFunctionCallContext functionCallContext = ctx.lastFunctionCall();
+ Path path = parseSuffixPath(functionCallContext.suffixPath());
+ selectOp.addLastPath(path, functionCallContext.LAST().getText());
+ queryOp.setSelectOperator(selectOp);
Review comment:
Sql syntax updated
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] wshao08 commented on a change in pull request
#821: [IOTDB-298]Last time-value query
Posted by GitBox <gi...@apache.org>.
wshao08 commented on a change in pull request #821: [IOTDB-298]Last time-value query
URL: https://github.com/apache/incubator-iotdb/pull/821#discussion_r380821805
##########
File path: server/src/main/java/org/apache/iotdb/db/qp/executor/PlanExecutor.java
##########
@@ -680,6 +674,7 @@ public void insert(InsertPlan insertPlan) throws QueryProcessException {
}
insertPlan.setDataTypes(dataTypes);
storageEngine.insert(insertPlan);
+ insertPlan.updateMNodeLastValues(node);
Review comment:
The update function is moved into MNode structure.
I added a loop to update the last value after storageEngine.insert() instead, since it is possible that insert() could fail and throw exceptions. In such case the last value cache should not be updated.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] wshao08 commented on a change in pull request
#821: [IOTDB-298]Last time-value query
Posted by GitBox <gi...@apache.org>.
wshao08 commented on a change in pull request #821: [IOTDB-298]Last time-value query
URL: https://github.com/apache/incubator-iotdb/pull/821#discussion_r386032623
##########
File path: server/src/main/java/org/apache/iotdb/db/metadata/mnode/LeafMNode.java
##########
@@ -74,4 +77,25 @@ public int getLeafCount() {
public MeasurementSchema getSchema() {
return schema;
}
+
+ public TimeValuePair getCachedLast() {
+ return cachedLastValuePair;
+ }
+
+ public synchronized void updateCachedLast(TimeValuePair timeValuePair, boolean insertionUpdate,
+ Long latestFlushedTime) {
+ if (timeValuePair == null || timeValuePair.getValue() == null) return;
Review comment:
Done
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] qiaojialin commented on a change in pull request
#821: [IOTDB-298]Last time-value query
Posted by GitBox <gi...@apache.org>.
qiaojialin commented on a change in pull request #821: [IOTDB-298]Last time-value query
URL: https://github.com/apache/incubator-iotdb/pull/821#discussion_r385680773
##########
File path: server/src/main/java/org/apache/iotdb/db/query/executor/LastQueryExecutor.java
##########
@@ -0,0 +1,202 @@
+/*
+ * 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.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);
+
+ for (int i = seqFileResources.size() - 1; i >= 0; i--) {
+ List<ChunkMetaData> chunkMetadata =
+ loadSatisfiedChunkMetadata(seqFileResources.get(i), seriesPath, context);
+ if (!chunkMetadata.isEmpty()) {
+ ChunkMetaData lastChunkMetaData = chunkMetadata.get(chunkMetadata.size() - 1);
+ Statistics chunkStatistics = lastChunkMetaData.getStatistics();
+ resultPair = constructLastPair(
+ chunkStatistics.getEndTime(), chunkStatistics.getLastValue(), tsDataType);
+ break;
+ }
+ }
+
+ long version = 0;
+ for (TsFileResource resource : unseqFileResources) {
+ if (resource.getEndTimeMap().get(seriesPath.getDevice()) < resultPair.getTimestamp()) {
+ break;
+ }
+ List<ChunkMetaData> chunkMetadata = loadSatisfiedChunkMetadata(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 List<ChunkMetaData> loadSatisfiedChunkMetadata(
Review comment:
extract this to a Tools static method
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] JackieTien97 commented on a change in pull
request #821: [IOTDB-298]Last time-value query
Posted by GitBox <gi...@apache.org>.
JackieTien97 commented on a change in pull request #821: [IOTDB-298]Last time-value query
URL: https://github.com/apache/incubator-iotdb/pull/821#discussion_r380541185
##########
File path: server/src/main/java/org/apache/iotdb/db/service/TSServiceImpl.java
##########
@@ -196,6 +196,7 @@ public static TSDataType getSeriesType(String path) throws QueryProcessException
case SQLConstant.FIRST_VALUE:
case SQLConstant.MIN_VALUE:
case SQLConstant.MAX_VALUE:
+ //case SQLConstant.LAST:
Review comment:
delete it
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] JackieTien97 commented on a change in pull
request #821: [IOTDB-298]Last time-value query
Posted by GitBox <gi...@apache.org>.
JackieTien97 commented on a change in pull request #821: [IOTDB-298]Last time-value query
URL: https://github.com/apache/incubator-iotdb/pull/821#discussion_r380540195
##########
File path: server/src/main/java/org/apache/iotdb/db/qp/physical/crud/BatchInsertPlan.java
##########
@@ -288,6 +292,23 @@ public long getMaxTime() {
return maxTime;
}
+ public void updateMNodeLastValues(MNode node) {
+ long maxTime = Long.MIN_VALUE;
+ int maxIndex = 0;
+ for (int i = 0; i < times.length; i++) {
+ if (times[i] > maxTime) {
+ maxTime = times[i];
+ maxIndex = i;
+ }
+ }
+ for (int i = 0; i < measurements.length; i++) {
+ if (node.hasChild(measurements[i])) {
+ Object[] column = (Object[]) columns[i];
+ node.getChild(measurements[i]).updateCachedLast(maxTime, column[maxIndex], dataTypes[i]);
+ }
+ }
+ }
+
Review comment:
This function is not the InsertPlan's responsibility
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] wshao08 commented on a change in pull request
#821: [IOTDB-298]Last time-value query
Posted by GitBox <gi...@apache.org>.
wshao08 commented on a change in pull request #821: [IOTDB-298]Last time-value query
URL: https://github.com/apache/incubator-iotdb/pull/821#discussion_r384646176
##########
File path: server/src/main/java/org/apache/iotdb/db/query/executor/LastQueryExecutor.java
##########
@@ -0,0 +1,216 @@
+/*
+ * 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 org.apache.iotdb.db.exception.StorageEngineException;
+import org.apache.iotdb.db.exception.path.PathException;
+import org.apache.iotdb.db.exception.query.QueryProcessException;
+import org.apache.iotdb.db.metadata.MManager;
+import org.apache.iotdb.db.metadata.MNode;
+import org.apache.iotdb.db.qp.physical.crud.LastQueryPlan;
+import org.apache.iotdb.db.query.aggregation.AggregateResult;
+import org.apache.iotdb.db.query.context.QueryContext;
+import org.apache.iotdb.db.query.control.QueryResourceManager;
+import org.apache.iotdb.db.query.dataset.SingleDataSet;
+import org.apache.iotdb.db.query.reader.series.IAggregateReader;
+import org.apache.iotdb.db.query.reader.series.SeriesAggregateReader;
+import org.apache.iotdb.tsfile.exception.cache.CacheException;
+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.common.BatchData;
+import org.apache.iotdb.tsfile.read.common.Path;
+import org.apache.iotdb.tsfile.read.common.RowRecord;
+import org.apache.iotdb.tsfile.read.filter.basic.Filter;
+import org.apache.iotdb.tsfile.read.query.dataset.QueryDataSet;
+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 {
+
+ List<LastQueryResult> lastQueryResultList = new ArrayList<>();
+ for (int i = 0; i < selectedSeries.size(); i++) {
+ LastQueryResult lastQueryResult = calculateLastPairForOneSeries(selectedSeries.get(i), dataTypes.get(i), context);
+ lastQueryResultList.add(lastQueryResult);
+ }
+
+ RowRecord resultRecord = constructLastRowRecord(lastQueryResultList);
+ SingleDataSet dataSet = new SingleDataSet(selectedSeries, dataTypes);
+ dataSet.setRecord(resultRecord);
+ return dataSet;
+ }
+
+ /**
+ * get aggregation result for one series
+ *
+ * @param context query context
+ * @return AggregateResult list
+ */
+ private LastQueryResult calculateLastPairForOneSeries(
+ Path seriesPath, TSDataType tsDataType,
+ QueryContext context)
+ throws IOException, QueryProcessException, StorageEngineException {
+ LastQueryResult queryResult = new LastQueryResult();
+ MNode node = null;
+ try {
+ node = MManager.getInstance().getNodeByPathFromCache(seriesPath.toString());
+ } catch (PathException e) {
+ throw new QueryProcessException(e);
+ } catch (CacheException e) {
+ throw new QueryProcessException(e.getMessage());
+ }
+ if (node.getCachedLast() != null) {
+ queryResult.setPairResult(node.getCachedLast());
+ return queryResult;
+ }
+
+ // construct series reader without value filter
+ Filter timeFilter = null;
+ IAggregateReader seriesReader = new SeriesAggregateReader(
+ seriesPath, tsDataType, context, QueryResourceManager.getInstance()
+ .getQueryDataSource(seriesPath, context, timeFilter), timeFilter, null);
+
+ long maxTime = Long.MIN_VALUE;
+ while (seriesReader.hasNextChunk()) {
+ // cal by chunk statistics
+ if (seriesReader.canUseCurrentChunkStatistics()) {
+ Statistics chunkStatistics = seriesReader.currentChunkStatistics();
+ if (chunkStatistics.getEndTime() > maxTime) {
+ maxTime = chunkStatistics.getEndTime();
+ queryResult.setPairResult(maxTime, chunkStatistics.getLastValue(), tsDataType);
+ }
+ seriesReader.skipCurrentChunk();
+ continue;
+ }
+ while (seriesReader.hasNextPage()) {
+ //cal by page statistics
+ if (seriesReader.canUseCurrentPageStatistics()) {
+ Statistics pageStatistic = seriesReader.currentPageStatistics();
+ if (pageStatistic.getEndTime() > maxTime) {
+ maxTime = pageStatistic.getEndTime();
+ queryResult.setPairResult(maxTime, pageStatistic.getLastValue(), tsDataType);
+ }
+ seriesReader.skipCurrentPage();
+ continue;
+ }
+ // cal by page data
+ while (seriesReader.hasNextOverlappedPage()) {
+ BatchData nextOverlappedPageData = seriesReader.nextOverlappedPage();
+ int maxIndex = nextOverlappedPageData.length() - 1;
+ if (maxIndex < 0) {
+ continue;
+ }
+ long time = nextOverlappedPageData.getTimeByIndex(maxIndex);
+ if (time > maxTime) {
+ maxTime = time;
+ queryResult.setPairResult(maxTime, nextOverlappedPageData.getValueInTimestamp(time), tsDataType);
+ }
+ nextOverlappedPageData.resetBatchData();
+ }
+ }
+ }
+ if (queryResult.hasResult())
+ node.setCachedLast(queryResult.getPairResult());
+ return queryResult;
+ }
+
+ /**
+ * using last result data list construct QueryDataSet.
+ *
+ * @param lastQueryResultList last result list
+ */
+ private RowRecord constructLastRowRecord(List<LastQueryResult> lastQueryResultList) {
+ long maxTime = Long.MIN_VALUE;
+ for (LastQueryResult lastPair : lastQueryResultList) {
+ if (lastPair.hasResult() && lastPair.getTimestamp() > maxTime)
+ maxTime = lastPair.getTimestamp();
+ }
+
+ RowRecord resultRecord = new RowRecord(maxTime);
+ for (int i = 0; i < lastQueryResultList.size(); i++) {
+ TSDataType dataType = dataTypes.get(i);
+ LastQueryResult lastPair = lastQueryResultList.get(i);
+ if (lastPair.hasResult() && lastPair.getTimestamp() == maxTime)
+ resultRecord.addField(lastPair.getValue(), dataType);
+ else
+ resultRecord.addField(null, dataType);
+ }
+
+ return resultRecord;
+ }
+
+ class LastQueryResult {
Review comment:
Removed in the new version
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] wshao08 commented on a change in pull request
#821: [IOTDB-298]Last time-value query
Posted by GitBox <gi...@apache.org>.
wshao08 commented on a change in pull request #821: [IOTDB-298]Last time-value query
URL: https://github.com/apache/incubator-iotdb/pull/821#discussion_r385593605
##########
File path: server/src/main/java/org/apache/iotdb/db/service/TSServiceImpl.java
##########
@@ -646,7 +644,7 @@ private TSExecuteStatementResp executeQueryStatement(
resp.setOperationType(plan.getOperatorType().toString());
// generate the queryId for the operation
long queryId = generateQueryId(true);
- // put it into the corresponding Set
+ // put it into the corresponding SetO
Review comment:
Fixed
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-iotdb] qiaojialin commented on a change in pull request
#821: [IOTDB-298]Last time-value query
Posted by GitBox <gi...@apache.org>.
qiaojialin commented on a change in pull request #821: [IOTDB-298]Last time-value query
URL: https://github.com/apache/incubator-iotdb/pull/821#discussion_r385672785
##########
File path: server/src/main/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessor.java
##########
@@ -496,15 +511,38 @@ public void insert(InsertPlan insertPlan) throws QueryProcessException {
latestFlushedTimeForEachDevice.computeIfAbsent(timePartitionId, id -> new HashMap<>())
.putIfAbsent(insertPlan.getDeviceId(), Long.MIN_VALUE);
+ long latestFlushedTime = globalLatestFlushedTimeForEachDevice.computeIfAbsent(
+ insertPlan.getDeviceId(), k -> Long.MIN_VALUE);
// insert to sequence or unSequence file
insertToTsFileProcessor(insertPlan,
insertPlan.getTime() > latestFlushedTimeForEachDevice.get(timePartitionId)
.get(insertPlan.getDeviceId()));
+
+ updateInsertPlanLast(insertPlan, latestFlushedTime);
+ if (latestFlushedTime < insertPlan.getTime())
+ globalLatestFlushedTimeForEachDevice.put(insertPlan.getDeviceId(), insertPlan.getTime());
} finally {
writeUnlock();
}
}
+ public void updateInsertPlanLast(InsertPlan plan, Long latestFlushedTime)
Review comment:
```suggestion
public void tryToUpdateCache(InsertPlan plan, Long latestFlushedTime)
```
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services