You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ja...@apache.org on 2023/05/16 11:45:35 UTC
[iotdb] branch caLastOpt updated (b9148b9029 -> c90289c449)
This is an automated email from the ASF dual-hosted git repository.
jackietien pushed a change to branch caLastOpt
in repository https://gitbox.apache.org/repos/asf/iotdb.git
from b9148b9029 Merge remote-tracking branch 'origin/master' into caLastOpt
new 89b4a2a57c Add RPC interface
new bff77d3697 cp
new c90289c449 wrong Update
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../lastCache/container/LastCacheContainer.java | 11 +-
.../iotdb/db/mpp/execution/driver/Driver.java | 21 ++-
.../last/AbstractUpdateLastCacheOperator.java | 13 ++
.../last/AlignedUpdateLastCacheOperator.java | 21 ++-
.../process/last/UpdateLastCacheOperator.java | 13 +-
.../mpp/metric/TimeSeriesMetadataCacheMetrics.java | 4 +-
.../db/mpp/plan/planner/OperatorTreeGenerator.java | 4 +
.../service/thrift/impl/ClientRPCServiceImpl.java | 187 +++++++++++++++++++++
thrift/src/main/thrift/client.thrift | 15 ++
9 files changed, 259 insertions(+), 30 deletions(-)
[iotdb] 02/03: cp
Posted by ja...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jackietien pushed a commit to branch caLastOpt
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit bff77d369795ac440fcf8fb3d98d3c29776abb57
Author: JackieTien97 <ja...@gmail.com>
AuthorDate: Tue May 16 19:43:23 2023 +0800
cp
---
.../last/AbstractUpdateLastCacheOperator.java | 13 +++++++++++++
.../last/AlignedUpdateLastCacheOperator.java | 21 ++++++++++-----------
.../process/last/UpdateLastCacheOperator.java | 13 ++++++-------
.../db/mpp/plan/planner/OperatorTreeGenerator.java | 4 ++++
4 files changed, 33 insertions(+), 18 deletions(-)
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/last/AbstractUpdateLastCacheOperator.java b/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/last/AbstractUpdateLastCacheOperator.java
index f0b1ffc49a..5ac2e13c05 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/last/AbstractUpdateLastCacheOperator.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/last/AbstractUpdateLastCacheOperator.java
@@ -19,18 +19,23 @@
package org.apache.iotdb.db.mpp.execution.operator.process.last;
+import org.apache.iotdb.commons.path.MeasurementPath;
import org.apache.iotdb.db.metadata.cache.DataNodeSchemaCache;
import org.apache.iotdb.db.mpp.execution.driver.DataDriverContext;
import org.apache.iotdb.db.mpp.execution.operator.Operator;
import org.apache.iotdb.db.mpp.execution.operator.OperatorContext;
import org.apache.iotdb.db.mpp.execution.operator.process.ProcessOperator;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.read.TimeValuePair;
import org.apache.iotdb.tsfile.read.common.block.TsBlock;
import org.apache.iotdb.tsfile.read.common.block.TsBlockBuilder;
+import org.apache.iotdb.tsfile.utils.TsPrimitiveType;
import com.google.common.collect.ImmutableList;
import com.google.common.util.concurrent.ListenableFuture;
+import javax.annotation.Nullable;
+
public abstract class AbstractUpdateLastCacheOperator implements ProcessOperator {
protected static final TsBlock LAST_QUERY_EMPTY_TSBLOCK =
new TsBlockBuilder(ImmutableList.of(TSDataType.TEXT, TSDataType.TEXT, TSDataType.TEXT))
@@ -80,6 +85,14 @@ public abstract class AbstractUpdateLastCacheOperator implements ProcessOperator
return databaseName;
}
+ protected void updateLastCache(
+ long time, @Nullable TsPrimitiveType value, MeasurementPath fullPath) {
+ if (needUpdateCache) {
+ TimeValuePair timeValuePair = new TimeValuePair(time, value);
+ lastCache.updateLastCache(getDatabaseName(), fullPath, timeValuePair, false, Long.MIN_VALUE);
+ }
+ }
+
@Override
public boolean hasNext() throws Exception {
return child.hasNextWithTimer();
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/last/AlignedUpdateLastCacheOperator.java b/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/last/AlignedUpdateLastCacheOperator.java
index a75581eb03..85a5900616 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/last/AlignedUpdateLastCacheOperator.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/last/AlignedUpdateLastCacheOperator.java
@@ -25,7 +25,6 @@ import org.apache.iotdb.commons.path.PartialPath;
import org.apache.iotdb.db.metadata.cache.DataNodeSchemaCache;
import org.apache.iotdb.db.mpp.execution.operator.Operator;
import org.apache.iotdb.db.mpp.execution.operator.OperatorContext;
-import org.apache.iotdb.tsfile.read.TimeValuePair;
import org.apache.iotdb.tsfile.read.common.block.TsBlock;
import org.apache.iotdb.tsfile.utils.TsPrimitiveType;
@@ -63,25 +62,25 @@ public class AlignedUpdateLastCacheOperator extends AbstractUpdateLastCacheOpera
tsBlockBuilder.reset();
for (int i = 0; i + 1 < res.getValueColumnCount(); i += 2) {
+ MeasurementPath measurementPath =
+ new MeasurementPath(
+ devicePath.concatNode(seriesPath.getMeasurementList().get(i / 2)),
+ seriesPath.getSchemaList().get(i / 2),
+ true);
if (!res.getColumn(i).isNull(0)) {
long lastTime = res.getColumn(i).getLong(0);
TsPrimitiveType lastValue = res.getColumn(i + 1).getTsPrimitiveType(0);
- MeasurementPath measurementPath =
- new MeasurementPath(
- devicePath.concatNode(seriesPath.getMeasurementList().get(i / 2)),
- seriesPath.getSchemaList().get(i / 2),
- true);
- if (needUpdateCache) {
- TimeValuePair timeValuePair = new TimeValuePair(lastTime, lastValue);
- lastCache.updateLastCache(
- getDatabaseName(), measurementPath, timeValuePair, false, Long.MIN_VALUE);
- }
+ updateLastCache(lastTime, lastValue, measurementPath);
LastQueryUtil.appendLastValue(
tsBlockBuilder,
lastTime,
measurementPath.getFullPath(),
lastValue.getStringValue(),
seriesPath.getSchemaList().get(i / 2).getType().name());
+ } else {
+ // we still need to update last cache if there is no data for this time series to avoid
+ // scanning all files each time
+ updateLastCache(Long.MIN_VALUE, null, measurementPath);
}
}
return !tsBlockBuilder.isEmpty() ? tsBlockBuilder.build() : LAST_QUERY_EMPTY_TSBLOCK;
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/last/UpdateLastCacheOperator.java b/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/last/UpdateLastCacheOperator.java
index 7bcc5a61e5..e45c18c019 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/last/UpdateLastCacheOperator.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/last/UpdateLastCacheOperator.java
@@ -23,7 +23,6 @@ import org.apache.iotdb.db.metadata.cache.DataNodeSchemaCache;
import org.apache.iotdb.db.mpp.execution.operator.Operator;
import org.apache.iotdb.db.mpp.execution.operator.OperatorContext;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
-import org.apache.iotdb.tsfile.read.TimeValuePair;
import org.apache.iotdb.tsfile.read.common.block.TsBlock;
import org.apache.iotdb.tsfile.utils.TsPrimitiveType;
@@ -34,10 +33,10 @@ public class UpdateLastCacheOperator extends AbstractUpdateLastCacheOperator {
// fullPath for queried time series
// It should be exact PartialPath, neither MeasurementPath nor AlignedPath, because lastCache only
// accept PartialPath
- private MeasurementPath fullPath;
+ private final MeasurementPath fullPath;
// dataType for queried time series;
- private String dataType;
+ private final String dataType;
public UpdateLastCacheOperator(
OperatorContext operatorContext,
@@ -65,16 +64,16 @@ public class UpdateLastCacheOperator extends AbstractUpdateLastCacheOperator {
// last value is null
if (res.getColumn(0).isNull(0)) {
+ // we still need to update last cache if there is no data for this time series to avoid
+ // scanning all files each time
+ updateLastCache(Long.MIN_VALUE, null, fullPath);
return LAST_QUERY_EMPTY_TSBLOCK;
}
long lastTime = res.getColumn(0).getLong(0);
TsPrimitiveType lastValue = res.getColumn(1).getTsPrimitiveType(0);
- if (needUpdateCache) {
- TimeValuePair timeValuePair = new TimeValuePair(lastTime, lastValue);
- lastCache.updateLastCache(getDatabaseName(), fullPath, timeValuePair, false, Long.MIN_VALUE);
- }
+ updateLastCache(lastTime, lastValue, fullPath);
tsBlockBuilder.reset();
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/OperatorTreeGenerator.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/OperatorTreeGenerator.java
index 4e945c8be6..e73bdd4236 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/OperatorTreeGenerator.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/OperatorTreeGenerator.java
@@ -2031,6 +2031,8 @@ public class OperatorTreeGenerator extends PlanVisitor<Operator, LocalExecutionP
TimeValuePair timeValuePair = DATA_NODE_SCHEMA_CACHE.getLastCache(seriesPath);
if (timeValuePair == null) { // last value is not cached
return createUpdateLastCacheOperator(node, context, node.getSeriesPath());
+ } else if (timeValuePair.getValue() == null) { // there is no data for this time series
+ return null;
} else if (!LastQueryUtil.satisfyFilter(
updateFilterUsingTTL(context.getLastQueryTimeFilter(), context.getDataRegionTTL()),
timeValuePair)) { // cached last value is not satisfied
@@ -2125,6 +2127,8 @@ public class OperatorTreeGenerator extends PlanVisitor<Operator, LocalExecutionP
TimeValuePair timeValuePair = DATA_NODE_SCHEMA_CACHE.getLastCache(measurementPath);
if (timeValuePair == null) { // last value is not cached
unCachedMeasurementIndexes.add(i);
+ } else if (timeValuePair.getValue() == null) {
+ // there is no data for this time series, just ignore
} else if (!LastQueryUtil.satisfyFilter(
updateFilterUsingTTL(context.getLastQueryTimeFilter(), context.getDataRegionTTL()),
timeValuePair)) { // cached last value is not satisfied
[iotdb] 03/03: wrong Update
Posted by ja...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jackietien pushed a commit to branch caLastOpt
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit c90289c449cf976af89b2ae026701fc0e0888540
Author: JackieTien97 <ja...@gmail.com>
AuthorDate: Tue May 16 19:45:26 2023 +0800
wrong Update
---
.../cache/lastCache/container/LastCacheContainer.java | 11 +++++++++--
1 file changed, 9 insertions(+), 2 deletions(-)
diff --git a/server/src/main/java/org/apache/iotdb/db/metadata/cache/lastCache/container/LastCacheContainer.java b/server/src/main/java/org/apache/iotdb/db/metadata/cache/lastCache/container/LastCacheContainer.java
index 65a2444f37..94057f3ca3 100644
--- a/server/src/main/java/org/apache/iotdb/db/metadata/cache/lastCache/container/LastCacheContainer.java
+++ b/server/src/main/java/org/apache/iotdb/db/metadata/cache/lastCache/container/LastCacheContainer.java
@@ -41,8 +41,15 @@ public class LastCacheContainer implements ILastCacheContainer {
@Override
public synchronized void updateCachedLast(
TimeValuePair timeValuePair, boolean highPriorityUpdate, Long latestFlushedTime) {
- if (timeValuePair == null || timeValuePair.getValue() == null) {
- return;
+ if (highPriorityUpdate) { // for write, we won't cache null value
+ if (timeValuePair == null || timeValuePair.getValue() == null) {
+ return;
+ }
+ } else { // for read, we need to cache null value, because it means that there is no data for
+ // this time series
+ if (timeValuePair == null) {
+ return;
+ }
}
if (lastCacheValue == null) {
[iotdb] 01/03: Add RPC interface
Posted by ja...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jackietien pushed a commit to branch caLastOpt
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 89b4a2a57c28d8fbafc918ac3837e7fe95fe0c2e
Author: JackieTien97 <ja...@gmail.com>
AuthorDate: Tue May 16 19:34:22 2023 +0800
Add RPC interface
---
.../iotdb/db/mpp/execution/driver/Driver.java | 21 ++-
.../mpp/metric/TimeSeriesMetadataCacheMetrics.java | 4 +-
.../service/thrift/impl/ClientRPCServiceImpl.java | 187 +++++++++++++++++++++
thrift/src/main/thrift/client.thrift | 15 ++
4 files changed, 217 insertions(+), 10 deletions(-)
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/execution/driver/Driver.java b/server/src/main/java/org/apache/iotdb/db/mpp/execution/driver/Driver.java
index 2ce330e976..b764a501b3 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/execution/driver/Driver.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/execution/driver/Driver.java
@@ -22,6 +22,7 @@ import org.apache.iotdb.commons.utils.FileUtils;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.mpp.execution.exchange.sink.ISink;
import org.apache.iotdb.db.mpp.execution.operator.Operator;
+import org.apache.iotdb.db.mpp.execution.operator.OperatorContext;
import org.apache.iotdb.db.mpp.execution.schedule.task.DriverTaskId;
import org.apache.iotdb.db.mpp.metric.QueryMetricsManager;
import org.apache.iotdb.tsfile.read.common.block.TsBlock;
@@ -36,7 +37,9 @@ import org.slf4j.LoggerFactory;
import javax.annotation.concurrent.GuardedBy;
import java.io.File;
+import java.util.HashMap;
import java.util.List;
+import java.util.Map;
import java.util.Optional;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
@@ -381,15 +384,17 @@ public abstract class Driver implements IDriver {
sink.setNoMoreTsBlocks();
+ Map<String, Long> operatorType2TotalCost = new HashMap<>();
// record operator execution statistics to metrics
- // List<OperatorContext> operatorContexts = driverContext.getOperatorContexts();
- // for (OperatorContext operatorContext : operatorContexts) {
- // String operatorType = operatorContext.getOperatorType();
- // QUERY_METRICS.recordOperatorExecutionCost(
- // operatorType, operatorContext.getTotalExecutionTimeInNanos());
- // QUERY_METRICS.recordOperatorExecutionCount(
- // operatorType, operatorContext.getNextCalledCount());
- // }
+ List<OperatorContext> operatorContexts = driverContext.getOperatorContexts();
+ for (OperatorContext operatorContext : operatorContexts) {
+ String operatorType = operatorContext.getOperatorType();
+ operatorType2TotalCost.merge(
+ operatorType, operatorContext.getTotalExecutionTimeInNanos(), Long::sum);
+ }
+ for (Map.Entry<String, Long> entry : operatorType2TotalCost.entrySet()) {
+ QUERY_METRICS.recordOperatorExecutionCost(entry.getKey(), entry.getValue());
+ }
} catch (InterruptedException t) {
// don't record the stack
wasInterrupted = true;
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/metric/TimeSeriesMetadataCacheMetrics.java b/server/src/main/java/org/apache/iotdb/db/mpp/metric/TimeSeriesMetadataCacheMetrics.java
index 965ed5483e..7a0ef0e4b9 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/metric/TimeSeriesMetadataCacheMetrics.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/metric/TimeSeriesMetadataCacheMetrics.java
@@ -43,14 +43,14 @@ public class TimeSeriesMetadataCacheMetrics implements IMetricSet {
Metric.CACHE_HIT.toString(),
MetricLevel.IMPORTANT,
timeSeriesMetadataCache,
- TimeSeriesMetadataCache::calculateTimeSeriesMetadataHitRatio,
+ h -> h.calculateTimeSeriesMetadataHitRatio() * 100,
Tag.NAME.toString(),
"timeSeriesMeta");
metricService.createAutoGauge(
Metric.CACHE_HIT.toString(),
MetricLevel.IMPORTANT,
timeSeriesMetadataCache,
- TimeSeriesMetadataCache::calculateBloomFilterHitRatio,
+ h -> h.calculateBloomFilterHitRatio() * 100,
Tag.NAME.toString(),
"bloomFilter");
}
diff --git a/server/src/main/java/org/apache/iotdb/db/service/thrift/impl/ClientRPCServiceImpl.java b/server/src/main/java/org/apache/iotdb/db/service/thrift/impl/ClientRPCServiceImpl.java
index 01bf923362..434acf1192 100644
--- a/server/src/main/java/org/apache/iotdb/db/service/thrift/impl/ClientRPCServiceImpl.java
+++ b/server/src/main/java/org/apache/iotdb/db/service/thrift/impl/ClientRPCServiceImpl.java
@@ -18,11 +18,15 @@
*/
package org.apache.iotdb.db.service.thrift.impl;
+import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet;
import org.apache.iotdb.common.rpc.thrift.TSStatus;
import org.apache.iotdb.commons.conf.CommonDescriptor;
import org.apache.iotdb.commons.conf.IoTDBConstant;
import org.apache.iotdb.commons.exception.IllegalPathException;
import org.apache.iotdb.commons.exception.IoTDBException;
+import org.apache.iotdb.commons.partition.DataPartition;
+import org.apache.iotdb.commons.partition.DataPartitionQueryParam;
+import org.apache.iotdb.commons.path.PartialPath;
import org.apache.iotdb.commons.service.metric.MetricService;
import org.apache.iotdb.commons.service.metric.enums.Metric;
import org.apache.iotdb.commons.service.metric.enums.Tag;
@@ -32,8 +36,11 @@ import org.apache.iotdb.db.auth.AuthorityChecker;
import org.apache.iotdb.db.conf.IoTDBConfig;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.conf.OperationType;
+import org.apache.iotdb.db.metadata.cache.DataNodeSchemaCache;
import org.apache.iotdb.db.metadata.template.TemplateQueryType;
import org.apache.iotdb.db.mpp.common.header.DatasetHeader;
+import org.apache.iotdb.db.mpp.common.header.DatasetHeaderFactory;
+import org.apache.iotdb.db.mpp.execution.operator.process.last.LastQueryUtil;
import org.apache.iotdb.db.mpp.plan.Coordinator;
import org.apache.iotdb.db.mpp.plan.analyze.ClusterPartitionFetcher;
import org.apache.iotdb.db.mpp.plan.analyze.IPartitionFetcher;
@@ -91,6 +98,7 @@ import org.apache.iotdb.service.rpc.thrift.TSDropSchemaTemplateReq;
import org.apache.iotdb.service.rpc.thrift.TSExecuteBatchStatementReq;
import org.apache.iotdb.service.rpc.thrift.TSExecuteStatementReq;
import org.apache.iotdb.service.rpc.thrift.TSExecuteStatementResp;
+import org.apache.iotdb.service.rpc.thrift.TSFastLastDataQueryForOneDeviceReq;
import org.apache.iotdb.service.rpc.thrift.TSFetchMetadataReq;
import org.apache.iotdb.service.rpc.thrift.TSFetchMetadataResp;
import org.apache.iotdb.service.rpc.thrift.TSFetchResultsReq;
@@ -118,8 +126,12 @@ import org.apache.iotdb.service.rpc.thrift.TSSetTimeZoneReq;
import org.apache.iotdb.service.rpc.thrift.TSUnsetSchemaTemplateReq;
import org.apache.iotdb.service.rpc.thrift.TSyncIdentityInfo;
import org.apache.iotdb.service.rpc.thrift.TSyncTransportMetaInfo;
+import org.apache.iotdb.tsfile.read.TimeValuePair;
import org.apache.iotdb.tsfile.read.common.block.TsBlock;
+import org.apache.iotdb.tsfile.read.common.block.TsBlockBuilder;
import org.apache.iotdb.tsfile.read.common.block.column.Column;
+import org.apache.iotdb.tsfile.read.common.block.column.TsBlockSerde;
+import org.apache.iotdb.tsfile.utils.Binary;
import org.apache.iotdb.tsfile.utils.Pair;
import org.apache.thrift.TException;
@@ -130,11 +142,13 @@ import java.io.IOException;
import java.nio.ByteBuffer;
import java.time.ZoneId;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.TimeUnit;
+import static org.apache.iotdb.db.mpp.common.DataNodeEndPoints.isSameNode;
import static org.apache.iotdb.db.utils.ErrorHandlingUtils.onIoTDBException;
import static org.apache.iotdb.db.utils.ErrorHandlingUtils.onNPEOrUnexpectedException;
import static org.apache.iotdb.db.utils.ErrorHandlingUtils.onQueryException;
@@ -158,6 +172,10 @@ public class ClientRPCServiceImpl implements IClientRPCServiceWithHandler {
private final ISchemaFetcher schemaFetcher;
+ private final TsBlockSerde serde = new TsBlockSerde();
+
+ private final DataNodeSchemaCache DATA_NODE_SCHEMA_CACHE = DataNodeSchemaCache.getInstance();
+
@FunctionalInterface
public interface SelectResult {
boolean apply(TSExecuteStatementResp resp, IQueryExecution queryExecution, int fetchSize)
@@ -552,6 +570,175 @@ public class ClientRPCServiceImpl implements IClientRPCServiceWithHandler {
return executeLastDataQueryInternal(req, SELECT_RESULT);
}
+ @Override
+ public TSExecuteStatementResp executeFastLastDataQueryForOneDeviceV2(
+ TSFastLastDataQueryForOneDeviceReq req) throws TException {
+ boolean finished = false;
+ long queryId = Long.MIN_VALUE;
+ IClientSession clientSession = SESSION_MANAGER.getCurrSessionAndUpdateIdleTime();
+ OperationQuota quota = null;
+ if (!SESSION_MANAGER.checkLogin(clientSession)) {
+ return RpcUtils.getTSExecuteStatementResp(getNotLoggedInStatus());
+ }
+ long startTime = System.currentTimeMillis();
+ long startNanoTime = System.nanoTime();
+ Throwable t = null;
+ try {
+ String db;
+ String deviceId;
+ PartialPath devicePath;
+
+ queryId = SESSION_MANAGER.requestQueryId(clientSession, req.statementId);
+
+ if (req.isLegalPathNodes()) {
+ db = req.db;
+ deviceId = req.deviceId;
+ devicePath = new PartialPath(deviceId.split("\\."));
+ } else {
+ db = new PartialPath(req.db).getFullPath();
+ devicePath = new PartialPath(req.deviceId);
+ deviceId = devicePath.getFullPath();
+ }
+
+ DataPartitionQueryParam queryParam =
+ new DataPartitionQueryParam(deviceId, Collections.emptyList(), true, true);
+ DataPartition dataPartition =
+ partitionFetcher.getDataPartitionWithUnclosedTimeRange(
+ Collections.singletonMap(db, Collections.singletonList(queryParam)));
+ List<TRegionReplicaSet> regionReplicaSets =
+ dataPartition.getDataRegionReplicaSet(deviceId, Collections.emptyList());
+ // if the device's latest dataRegion's leader is current node, we can directly read from cache
+ if (!regionReplicaSets.isEmpty()
+ && isSameNode(
+ regionReplicaSets
+ .get(regionReplicaSets.size() - 1)
+ .dataNodeLocations
+ .get(0)
+ .mPPDataExchangeEndPoint)) {
+ int sensorNum = req.sensors.size();
+ TsBlockBuilder builder = LastQueryUtil.createTsBlockBuilder(sensorNum);
+ boolean allCached = true;
+ for (String sensor : req.sensors) {
+ PartialPath fullPath;
+ if (req.isLegalPathNodes()) {
+ fullPath = devicePath.concatNode(sensor);
+ } else {
+ fullPath = devicePath.concatNode((new PartialPath(sensor)).getFullPath());
+ }
+ TimeValuePair timeValuePair = DATA_NODE_SCHEMA_CACHE.getLastCache(fullPath);
+ if (timeValuePair == null) {
+ allCached = false;
+ break;
+ } else if (timeValuePair.getValue() == null) {
+ // there is no data for this sensor
+ } else {
+ // we don't consider TTL
+ LastQueryUtil.appendLastValue(
+ builder,
+ timeValuePair.getTimestamp(),
+ new Binary(fullPath.getFullPath()),
+ timeValuePair.getValue().getStringValue(),
+ timeValuePair.getValue().getDataType().name());
+ }
+ }
+ // cache hit
+ if (allCached) {
+ TSExecuteStatementResp resp =
+ createResponse(DatasetHeaderFactory.getLastQueryHeader(), queryId);
+ resp.setStatus(RpcUtils.getStatus(TSStatusCode.SUCCESS_STATUS, ""));
+ resp.setQueryResult(Collections.singletonList(serde.serialize(builder.build())));
+ finished = true;
+ resp.setMoreData(true);
+ return resp;
+ }
+ }
+
+ // cache miss
+ Statement s = StatementGenerator.createStatement(convert(req), clientSession.getZoneId());
+ // permission check
+ TSStatus status = AuthorityChecker.checkAuthority(s, clientSession);
+ if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ return RpcUtils.getTSExecuteStatementResp(status);
+ }
+
+ quota =
+ DataNodeThrottleQuotaManager.getInstance()
+ .checkQuota(SESSION_MANAGER.getCurrSession().getUsername(), s);
+
+ if (enableAuditLog) {
+ AuditLogger.log(String.format("Last Data Query: %s", req), s);
+ }
+ // create and cache dataset
+ ExecutionResult result =
+ COORDINATOR.execute(
+ s,
+ queryId,
+ SESSION_MANAGER.getSessionInfo(clientSession),
+ "",
+ partitionFetcher,
+ schemaFetcher,
+ req.getTimeout());
+
+ if (result.status.code != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ throw new RuntimeException("error code: " + result.status);
+ }
+
+ IQueryExecution queryExecution = COORDINATOR.getQueryExecution(queryId);
+
+ try (SetThreadName threadName = new SetThreadName(result.queryId.getId())) {
+ TSExecuteStatementResp resp;
+ if (queryExecution.isQuery()) {
+ resp = createResponse(queryExecution.getDatasetHeader(), queryId);
+ resp.setStatus(result.status);
+ finished = SELECT_RESULT.apply(resp, queryExecution, req.fetchSize);
+ resp.setMoreData(!finished);
+ quota.addReadResult(resp.getQueryResult());
+ } else {
+ resp = RpcUtils.getTSExecuteStatementResp(result.status);
+ }
+ return resp;
+ }
+
+ } catch (Exception e) {
+ finished = true;
+ t = e;
+ return RpcUtils.getTSExecuteStatementResp(
+ onQueryException(e, "\"" + req + "\". " + OperationType.EXECUTE_LAST_DATA_QUERY));
+ } catch (Error error) {
+ t = error;
+ throw error;
+ } finally {
+ COORDINATOR.recordExecutionTime(queryId, System.currentTimeMillis() - startTime);
+ if (finished) {
+ addStatementExecutionLatency(
+ OperationType.EXECUTE_LAST_DATA_QUERY,
+ StatementType.QUERY,
+ System.nanoTime() - startNanoTime,
+ TimeUnit.NANOSECONDS);
+ COORDINATOR.cleanupQueryExecution(queryId, t);
+ }
+ SESSION_MANAGER.updateIdleTime();
+ if (quota != null) {
+ quota.close();
+ }
+ }
+ }
+
+ private TSLastDataQueryReq convert(TSFastLastDataQueryForOneDeviceReq req) {
+ TSLastDataQueryReq res = new TSLastDataQueryReq();
+ List<String> paths = new ArrayList<>(req.sensors);
+ for (String sensor : req.sensors) {
+ paths.add(req.deviceId + "." + sensor);
+ }
+ TSLastDataQueryReq tsLastDataQueryReq =
+ new TSLastDataQueryReq(req.sessionId, paths, 0, req.statementId);
+ tsLastDataQueryReq.setFetchSize(req.fetchSize);
+ tsLastDataQueryReq.setEnableRedirectQuery(req.enableRedirectQuery);
+ tsLastDataQueryReq.setLegalPathNodes(req.legalPathNodes);
+ tsLastDataQueryReq.setTimeout(req.timeout);
+ return res;
+ }
+
@Override
public TSExecuteStatementResp executeAggregationQueryV2(TSAggregationQueryReq req) {
return executeAggregationQueryInternal(req, SELECT_RESULT);
diff --git a/thrift/src/main/thrift/client.thrift b/thrift/src/main/thrift/client.thrift
index 6779bbf89c..6e0f3fcb99 100644
--- a/thrift/src/main/thrift/client.thrift
+++ b/thrift/src/main/thrift/client.thrift
@@ -338,6 +338,19 @@ struct TSLastDataQueryReq {
9: optional bool legalPathNodes
}
+struct TSFastLastDataQueryForOneDeviceReq {
+ 1: required i64 sessionId
+ 2: required string db
+ 3: required string deviceId
+ 4: required list<string> sensors
+ 5: optional i32 fetchSize
+ 6: required i64 statementId
+ 7: optional bool enableRedirectQuery
+ 8: optional bool jdbcQuery
+ 9: optional i64 timeout
+ 10: optional bool legalPathNodes
+}
+
struct TSAggregationQueryReq {
1: required i64 sessionId
2: required i64 statementId
@@ -491,6 +504,8 @@ service IClientRPCService {
TSExecuteStatementResp executeLastDataQueryV2(1:TSLastDataQueryReq req);
+ TSExecuteStatementResp executeFastLastDataQueryForOneDeviceV2(1:TSFastLastDataQueryForOneDeviceReq req);
+
TSExecuteStatementResp executeAggregationQueryV2(1:TSAggregationQueryReq req);
TSFetchResultsResp fetchResultsV2(1:TSFetchResultsReq req);