Skip to content

Commit a5c3c2e

Browse files
CaideyipiCaideyipi
committed
Fast last query on local
1 parent c58680b commit a5c3c2e

17 files changed

Lines changed: 284 additions & 6 deletions

File tree

‎iotdb-client/isession/src/main/java/org/apache/iotdb/isession/ISession.java‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,7 @@
2525
import org.apache.iotdb.isession.util.SystemStatus;
2626
import org.apache.iotdb.isession.util.Version;
2727
import org.apache.iotdb.rpc.IoTDBConnectionException;
28+
import org.apache.iotdb.rpc.RedirectException;
2829
import org.apache.iotdb.rpc.StatementExecutionException;
2930
import org.apache.iotdb.service.rpc.thrift.TSBackupConfigurationResp;
3031
import org.apache.iotdb.service.rpc.thrift.TSConnectionInfoResp;
@@ -185,6 +186,9 @@ SessionDataSet executeLastDataQuery(List<String> paths, long lastTime, long time
185186
SessionDataSet executeLastDataQuery(List<String> paths)
186187
throws StatementExecutionException, IoTDBConnectionException;
187188

189+
SessionDataSet executeFastLastDataQueryForOnePrefixPath(final List<String> prefixes)
190+
throws IoTDBConnectionException, StatementExecutionException, RedirectException;
191+
188192
SessionDataSet executeLastDataQueryForOneDevice(
189193
String db, String device, List<String> sensors, boolean isLegalPathNodes)
190194
throws StatementExecutionException, IoTDBConnectionException;

‎iotdb-client/isession/src/main/java/org/apache/iotdb/isession/pool/ISessionPool.java‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,7 @@
2424
import org.apache.iotdb.isession.util.SystemStatus;
2525
import org.apache.iotdb.isession.util.Version;
2626
import org.apache.iotdb.rpc.IoTDBConnectionException;
27+
import org.apache.iotdb.rpc.RedirectException;
2728
import org.apache.iotdb.rpc.StatementExecutionException;
2829
import org.apache.iotdb.service.rpc.thrift.TSBackupConfigurationResp;
2930
import org.apache.iotdb.service.rpc.thrift.TSConnectionInfoResp;
@@ -486,6 +487,9 @@ SessionDataSetWrapper executeLastDataQueryForOneDevice(
486487
String db, String device, List<String> sensors, boolean isLegalPathNodes)
487488
throws StatementExecutionException, IoTDBConnectionException;
488489

490+
SessionDataSetWrapper executeFastLastDataQueryForOnePrefixPath(final List<String> prefixes)
491+
throws IoTDBConnectionException, StatementExecutionException, RedirectException;
492+
489493
SessionDataSetWrapper executeAggregationQuery(
490494
List<String> paths, List<TAggregationType> aggregations)
491495
throws StatementExecutionException, IoTDBConnectionException;

‎iotdb-client/session/src/main/java/org/apache/iotdb/session/Session.java‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1105,6 +1105,12 @@ public SessionDataSet executeLastDataQuery(List<String> paths)
11051105
return executeLastDataQuery(paths, time, queryTimeoutInMs);
11061106
}
11071107

1108+
@Override
1109+
public SessionDataSet executeFastLastDataQueryForOnePrefixPath(final List<String> prefixes)
1110+
throws IoTDBConnectionException, StatementExecutionException, RedirectException {
1111+
return defaultSessionConnection.executeLastDataQueryForOnePrefixPath(prefixes);
1112+
}
1113+
11081114
@Override
11091115
public SessionDataSet executeLastDataQueryForOneDevice(
11101116
String db, String device, List<String> sensors, boolean isLegalPathNodes)

‎iotdb-client/session/src/main/java/org/apache/iotdb/session/SessionConnection.java‎

Lines changed: 39 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -46,6 +46,7 @@
4646
import org.apache.iotdb.service.rpc.thrift.TSExecuteStatementReq;
4747
import org.apache.iotdb.service.rpc.thrift.TSExecuteStatementResp;
4848
import org.apache.iotdb.service.rpc.thrift.TSFastLastDataQueryForOneDeviceReq;
49+
import org.apache.iotdb.service.rpc.thrift.TSFastLastDataQueryForOnePrefixPathReq;
4950
import org.apache.iotdb.service.rpc.thrift.TSInsertRecordReq;
5051
import org.apache.iotdb.service.rpc.thrift.TSInsertRecordsOfOneDeviceReq;
5152
import org.apache.iotdb.service.rpc.thrift.TSInsertRecordsReq;
@@ -495,6 +496,43 @@ protected SessionDataSet executeRawDataQuery(
495496
execResp.getColumnIndex2TsBlockColumnIndexList());
496497
}
497498

499+
protected SessionDataSet executeLastDataQueryForOnePrefixPath(final List<String> prefixes)
500+
throws StatementExecutionException, IoTDBConnectionException, RedirectException {
501+
TSFastLastDataQueryForOnePrefixPathReq req =
502+
new TSFastLastDataQueryForOnePrefixPathReq(sessionId, prefixes, statementId);
503+
req.setFetchSize(session.fetchSize);
504+
req.setEnableRedirectQuery(enableRedirect);
505+
506+
RetryResult<TSExecuteStatementResp> result =
507+
callWithReconnect(
508+
() -> {
509+
req.setSessionId(sessionId);
510+
req.setStatementId(statementId);
511+
return client.executeFastLastDataQueryForOnePrefixPath(req);
512+
});
513+
final TSExecuteStatementResp tsExecuteStatementResp = result.getResult();
514+
515+
if (result.getRetryAttempts() == 0) {
516+
RpcUtils.verifySuccessWithRedirection(tsExecuteStatementResp.getStatus());
517+
} else {
518+
RpcUtils.verifySuccess(tsExecuteStatementResp.getStatus());
519+
}
520+
521+
return new SessionDataSet(
522+
"",
523+
tsExecuteStatementResp.getColumns(),
524+
tsExecuteStatementResp.getDataTypeList(),
525+
tsExecuteStatementResp.columnNameIndexMap,
526+
tsExecuteStatementResp.getQueryId(),
527+
statementId,
528+
client,
529+
sessionId,
530+
tsExecuteStatementResp.queryResult,
531+
tsExecuteStatementResp.isIgnoreTimeStamp(),
532+
tsExecuteStatementResp.moreData,
533+
zoneId);
534+
}
535+
498536
protected Pair<SessionDataSet, TEndPoint> executeLastDataQueryForOneDevice(
499537
String db, String device, List<String> sensors, boolean isLegalPathNodes, long timeOut)
500538
throws StatementExecutionException, IoTDBConnectionException {
@@ -550,7 +588,7 @@ protected SessionDataSet executeLastDataQuery(List<String> paths, long time, lon
550588
throws StatementExecutionException, IoTDBConnectionException, RedirectException {
551589
TSLastDataQueryReq tsLastDataQueryReq =
552590
new TSLastDataQueryReq(sessionId, paths, time, statementId);
553-
tsLastDataQueryReq.setFetchSize(session.fetchSize);
591+
tsLastDataQueryReq.setFetchSize(60000);
554592
tsLastDataQueryReq.setEnableRedirectQuery(enableRedirect);
555593
tsLastDataQueryReq.setTimeout(timeOut);
556594

‎iotdb-client/session/src/main/java/org/apache/iotdb/session/pool/SessionPool.java‎

Lines changed: 27 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3214,6 +3214,33 @@ public SessionDataSetWrapper executeLastDataQuery(List<String> paths)
32143214
return null;
32153215
}
32163216

3217+
@Override
3218+
public SessionDataSetWrapper executeFastLastDataQueryForOnePrefixPath(final List<String> prefixes)
3219+
throws IoTDBConnectionException, StatementExecutionException {
3220+
for (int i = 0; i < RETRY; i++) {
3221+
ISession session = getSession();
3222+
try {
3223+
SessionDataSet resp = session.executeFastLastDataQueryForOnePrefixPath(prefixes);
3224+
SessionDataSetWrapper wrapper = new SessionDataSetWrapper(resp, session, this);
3225+
occupy(session);
3226+
return wrapper;
3227+
} catch (IoTDBConnectionException e) {
3228+
// TException means the connection is broken, remove it and get a new one.
3229+
LOGGER.warn("executeLastDataQuery failed", e);
3230+
cleanSessionAndMayThrowConnectionException(session, i, e);
3231+
} catch (StatementExecutionException | RuntimeException e) {
3232+
putBack(session);
3233+
throw e;
3234+
} catch (Throwable e) {
3235+
LOGGER.error(EXECUTE_LASTDATAQUERY_ERROR, e);
3236+
putBack(session);
3237+
throw new RuntimeException(e);
3238+
}
3239+
}
3240+
// never go here
3241+
return null;
3242+
}
3243+
32173244
@Override
32183245
public SessionDataSetWrapper executeLastDataQueryForOneDevice(
32193246
String db, String device, List<String> sensors, boolean isLegalPathNodes)

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/ClientRPCServiceImpl.java‎

Lines changed: 86 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -89,6 +89,8 @@
8989
import org.apache.iotdb.db.queryengine.plan.planner.plan.parameter.InputLocation;
9090
import org.apache.iotdb.db.queryengine.plan.planner.plan.parameter.SeriesScanOptions;
9191
import org.apache.iotdb.db.queryengine.plan.relational.metadata.Metadata;
92+
import org.apache.iotdb.db.queryengine.plan.relational.metadata.fetcher.cache.TableDeviceSchemaCache;
93+
import org.apache.iotdb.db.queryengine.plan.relational.metadata.fetcher.cache.TableId;
9294
import org.apache.iotdb.db.queryengine.plan.relational.metadata.fetcher.cache.TreeDeviceSchemaCacheManager;
9395
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.SetSqlDialect;
9496
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.Use;
@@ -116,6 +118,8 @@
116118
import org.apache.iotdb.db.queryengine.plan.statement.metadata.template.UnsetSchemaTemplateStatement;
117119
import org.apache.iotdb.db.queryengine.plan.statement.metadata.view.CreateTableViewStatement;
118120
import org.apache.iotdb.db.queryengine.plan.statement.sys.SetSqlDialectStatement;
121+
import org.apache.iotdb.db.schemaengine.SchemaEngine;
122+
import org.apache.iotdb.db.schemaengine.schemaregion.ISchemaRegion;
119123
import org.apache.iotdb.db.schemaengine.template.TemplateQueryType;
120124
import org.apache.iotdb.db.storageengine.StorageEngine;
121125
import org.apache.iotdb.db.storageengine.dataregion.DataRegion;
@@ -151,6 +155,7 @@
151155
import org.apache.iotdb.service.rpc.thrift.TSExecuteStatementReq;
152156
import org.apache.iotdb.service.rpc.thrift.TSExecuteStatementResp;
153157
import org.apache.iotdb.service.rpc.thrift.TSFastLastDataQueryForOneDeviceReq;
158+
import org.apache.iotdb.service.rpc.thrift.TSFastLastDataQueryForOnePrefixPathReq;
154159
import org.apache.iotdb.service.rpc.thrift.TSFetchMetadataReq;
155160
import org.apache.iotdb.service.rpc.thrift.TSFetchMetadataResp;
156161
import org.apache.iotdb.service.rpc.thrift.TSFetchResultsReq;
@@ -187,6 +192,7 @@
187192
import org.apache.tsfile.block.column.Column;
188193
import org.apache.tsfile.common.conf.TSFileConfig;
189194
import org.apache.tsfile.common.conf.TSFileDescriptor;
195+
import org.apache.tsfile.common.constant.TsFileConstant;
190196
import org.apache.tsfile.enums.TSDataType;
191197
import org.apache.tsfile.file.metadata.IDeviceID;
192198
import org.apache.tsfile.file.metadata.IDeviceID.Factory;
@@ -926,6 +932,86 @@ public TSExecuteStatementResp executeLastDataQueryV2(TSLastDataQueryReq req) {
926932
return executeLastDataQueryInternal(req, SELECT_RESULT);
927933
}
928934

935+
@Override
936+
public TSExecuteStatementResp executeFastLastDataQueryForOnePrefixPath(
937+
final TSFastLastDataQueryForOnePrefixPathReq req) {
938+
final IClientSession clientSession = SESSION_MANAGER.getCurrSessionAndUpdateIdleTime();
939+
if (!SESSION_MANAGER.checkLogin(clientSession)) {
940+
return RpcUtils.getTSExecuteStatementResp(getNotLoggedInStatus());
941+
}
942+
943+
try {
944+
final long queryId = SESSION_MANAGER.requestQueryId(clientSession, req.statementId);
945+
// 1. Map<Device, String[] measurements> ISchemaFetcher.getAllSensors(prefix) ~= 50ms
946+
947+
final PartialPath prefixPath = new PartialPath(req.getPrefixes().toArray(new String[0]));
948+
final Map<TableId, Map<IDeviceID, Map<String, TimeValuePair>>> resultMap = new HashMap<>();
949+
int sensorNum = 0;
950+
951+
for (final ISchemaRegion region : SchemaEngine.getInstance().getAllSchemaRegions()) {
952+
// Do not filter by database, since we assume all the regions match the prefix path
953+
sensorNum += region.fillLastQueryMap(prefixPath, resultMap);
954+
}
955+
956+
// 2.DATA_NODE_SCHEMA_CACHE.getLastCache()
957+
if (!TableDeviceSchemaCache.getInstance().getLastCache(resultMap)) {
958+
// 2.1 any sensor miss cache, construct last query sql, then return
959+
return executeLastDataQueryInternal(convert(req), SELECT_RESULT);
960+
}
961+
962+
// 2.2 all sensors hit cache, return response ~= 20ms
963+
final TsBlockBuilder builder = LastQueryUtil.createTsBlockBuilder(sensorNum);
964+
965+
for (final Map.Entry<TableId, Map<IDeviceID, Map<String, TimeValuePair>>> result :
966+
resultMap.entrySet()) {
967+
for (final Map.Entry<IDeviceID, Map<String, TimeValuePair>> device2MeasurementLastEntry :
968+
result.getValue().entrySet()) {
969+
final String deviceWithSeparator =
970+
device2MeasurementLastEntry.getKey().toString() + TsFileConstant.PATH_SEPARATOR;
971+
for (final Map.Entry<String, TimeValuePair> measurementLastEntry :
972+
device2MeasurementLastEntry.getValue().entrySet()) {
973+
final TimeValuePair tvPair = measurementLastEntry.getValue();
974+
LastQueryUtil.appendLastValue(
975+
builder,
976+
tvPair.getTimestamp(),
977+
deviceWithSeparator + measurementLastEntry.getKey(),
978+
tvPair.getValue().getStringValue(),
979+
tvPair.getValue().getDataType().name());
980+
}
981+
}
982+
}
983+
984+
final TSExecuteStatementResp resp =
985+
createResponse(DatasetHeaderFactory.getLastQueryHeader(), queryId);
986+
resp.setStatus(RpcUtils.getStatus(TSStatusCode.SUCCESS_STATUS, ""));
987+
if (builder.isEmpty()) {
988+
resp.setQueryResult(Collections.emptyList());
989+
} else {
990+
resp.setQueryResult(Collections.singletonList(serde.serialize(builder.build())));
991+
}
992+
993+
resp.setMoreData(false);
994+
return resp;
995+
} catch (final Exception e) {
996+
return RpcUtils.getTSExecuteStatementResp(
997+
onQueryException(e, "\"" + req + "\". " + OperationType.EXECUTE_LAST_DATA_QUERY));
998+
}
999+
}
1000+
1001+
private TSLastDataQueryReq convert(final TSFastLastDataQueryForOnePrefixPathReq req) {
1002+
TSLastDataQueryReq tsLastDataQueryReq =
1003+
new TSLastDataQueryReq(
1004+
req.sessionId,
1005+
Collections.singletonList(String.join(".", req.getPrefixes()) + ".**"),
1006+
Long.MIN_VALUE,
1007+
req.statementId);
1008+
tsLastDataQueryReq.setFetchSize(60000);
1009+
tsLastDataQueryReq.setEnableRedirectQuery(req.enableRedirectQuery);
1010+
tsLastDataQueryReq.setLegalPathNodes(true);
1011+
tsLastDataQueryReq.setTimeout(req.timeout);
1012+
return tsLastDataQueryReq;
1013+
}
1014+
9291015
@Override
9301016
public TSExecuteStatementResp executeFastLastDataQueryForOneDeviceV2(
9311017
TSFastLastDataQueryForOneDeviceReq req) {

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/cache/schema/dualkeycache/IDualKeyCache.java‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,8 @@
2121

2222
import javax.annotation.concurrent.GuardedBy;
2323

24+
import java.util.Map;
25+
import java.util.function.BiFunction;
2426
import java.util.function.Predicate;
2527
import java.util.function.ToIntFunction;
2628

@@ -37,6 +39,9 @@ public interface IDualKeyCache<FK, SK, V> {
3739
/** Get the cache value with given first key and second key. */
3840
V get(final FK firstKey, final SK secondKey);
3941

42+
<R> boolean batchApply(
43+
final Map<FK, Map<SK, R>> inputMap, final BiFunction<V, R, Boolean> mappingFunction);
44+
4045
/**
4146
* Update the existing value. The updater shall return the difference caused by the update,
4247
* because we do not want to call "valueSizeComputer" twice, which may include abundant useless

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/cache/schema/dualkeycache/impl/DualKeyCacheImpl.java‎

Lines changed: 22 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -33,6 +33,7 @@
3333
import java.util.Map;
3434
import java.util.Objects;
3535
import java.util.concurrent.ConcurrentHashMap;
36+
import java.util.function.BiFunction;
3637
import java.util.function.Predicate;
3738
import java.util.function.ToIntFunction;
3839

@@ -76,14 +77,33 @@ public V get(final FK firstKey, final SK secondKey) {
7677
}
7778
}
7879

80+
public <R> boolean batchApply(
81+
final Map<FK, Map<SK, R>> inputMap, final BiFunction<V, R, Boolean> mappingFunction) {
82+
for (final Map.Entry<FK, Map<SK, R>> fkMapEntry : inputMap.entrySet()) {
83+
final ICacheEntryGroup<FK, SK, V, T> cacheEntryGroup = firstKeyMap.get(fkMapEntry.getKey());
84+
if (cacheEntryGroup == null) {
85+
return false;
86+
}
87+
for (final Map.Entry<SK, R> skrEntry : fkMapEntry.getValue().entrySet()) {
88+
final T cacheEntry = cacheEntryGroup.getCacheEntry(skrEntry.getKey());
89+
if (cacheEntry == null) {
90+
return false;
91+
}
92+
if (!mappingFunction.apply(cacheEntry.getValue(), skrEntry.getValue())) {
93+
return false;
94+
}
95+
}
96+
}
97+
return true;
98+
}
99+
79100
@Override
80101
public void update(
81102
final FK firstKey,
82103
final @Nonnull SK secondKey,
83104
final V value,
84105
final ToIntFunction<V> updater,
85106
final boolean createIfNotExists) {
86-
87107
ICacheEntryGroup<FK, SK, V, T> cacheEntryGroup = firstKeyMap.get(firstKey);
88108
if (Objects.isNull(cacheEntryGroup)) {
89109
if (createIfNotExists) {
@@ -329,7 +349,7 @@ private long getEntriesCount() {
329349
private static class SegmentedConcurrentHashMap<K, V> {
330350

331351
private static final int SLOT_NUM = 31;
332-
352+
private int size = 0;
333353
private final Map<K, V>[] maps = new ConcurrentHashMap[SLOT_NUM];
334354

335355
V get(final K key) {

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/cache/SchemaCacheEntry.java‎

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -52,7 +52,6 @@ public class SchemaCacheEntry implements IMeasurementSchemaInfo {
5252
(int) RamUsageEstimator.shallowSizeOfInstance(SchemaCacheEntry.class) + 75;
5353

5454
private final IMeasurementSchema iMeasurementSchema;
55-
5655
private final Map<String, String> tagMap;
5756

5857
public SchemaCacheEntry(

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/cache/TableDeviceCacheEntry.java‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -217,6 +217,18 @@ TimeValuePair getTimeValuePair(final String measurement) {
217217
return Objects.nonNull(cache) ? cache.getTimeValuePair(measurement) : null;
218218
}
219219

220+
boolean updateInputMap(final @Nonnull Map<String, TimeValuePair> updateMap) {
221+
// Shall only call this for original table device
222+
for (final String measurement : updateMap.keySet()) {
223+
final TimeValuePair result = getTimeValuePair(measurement);
224+
if (result == null) {
225+
return false;
226+
}
227+
updateMap.put(measurement, result);
228+
}
229+
return true;
230+
}
231+
220232
// Shall pass in "" if last by time
221233
Optional<Pair<OptionalLong, TsPrimitiveType[]>> getLastRow(
222234
final String sourceMeasurement, final List<String> targetMeasurements) {

0 commit comments

Comments
 (0)