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 2025/08/21 12:46:36 UTC
(iotdb) 01/01: Support print fast last query request in sample query log
This is an automated email from the ASF dual-hosted git repository.
jackietien pushed a commit to branch RestAPI
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 0cb85aca1d7a10d30268adf0f48567e90a6e2d67
Author: JackieTien97 <ja...@gmail.com>
AuthorDate: Thu Aug 21 20:46:17 2025 +0800
Support print fast last query request in sample query log
---
.../iotdb/db/conf/rest/IoTDBRestServiceConfig.java | 2 +-
.../protocol/rest/v2/impl/RestApiServiceImpl.java | 43 +++++++++++---
.../protocol/thrift/impl/ClientRPCServiceImpl.java | 10 ++++
.../db/queryengine/common/MPPQueryContext.java | 7 +--
.../iotdb/db/queryengine/plan/Coordinator.java | 69 +++++++++++++++-------
5 files changed, 96 insertions(+), 35 deletions(-)
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/rest/IoTDBRestServiceConfig.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/rest/IoTDBRestServiceConfig.java
index 64c0f65fe30..12f70f9ce31 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/rest/IoTDBRestServiceConfig.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/rest/IoTDBRestServiceConfig.java
@@ -22,7 +22,7 @@ package org.apache.iotdb.db.conf.rest;
public class IoTDBRestServiceConfig {
/** If the enableRestService is true, we will start REST Service. */
- private boolean enableRestService = false;
+ private boolean enableRestService = true;
/** Set the REST Service port. */
private int restServicePort = 18080;
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/rest/v2/impl/RestApiServiceImpl.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/rest/v2/impl/RestApiServiceImpl.java
index 1dfb3da8c45..7490d0a79b9 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/rest/v2/impl/RestApiServiceImpl.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/rest/v2/impl/RestApiServiceImpl.java
@@ -78,15 +78,20 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
+import java.util.function.Supplier;
+
+import static org.apache.iotdb.db.queryengine.plan.Coordinator.recordQueries;
public class RestApiServiceImpl extends RestApiService {
- private static final IoTDBConfig config = IoTDBDescriptor.getInstance().getConfig();
+ private static final IoTDBConfig CONFIG = IoTDBDescriptor.getInstance().getConfig();
private static final Coordinator COORDINATOR = Coordinator.getInstance();
private static final SessionManager SESSION_MANAGER = SessionManager.getInstance();
+ private static final String FORMAT = "rest/v2/fastLastQuery %s";
+
private final IPartitionFetcher partitionFetcher;
private final ISchemaFetcher schemaFetcher;
@@ -109,6 +114,7 @@ public class RestApiServiceImpl extends RestApiService {
Statement statement = null;
boolean finish = false;
long startTime = System.nanoTime();
+ Throwable t = null;
try {
RequestValidationHandler.validatePrefixPaths(prefixPathList);
@@ -149,10 +155,10 @@ public class RestApiServiceImpl extends RestApiService {
statement,
queryId,
sessionInfo,
- "",
+ restFastLastQuerySql(prefixPathList),
partitionFetcher,
schemaFetcher,
- config.getQueryTimeoutThreshold(),
+ CONFIG.getQueryTimeoutThreshold(),
true);
finish = true;
@@ -203,13 +209,14 @@ public class RestApiServiceImpl extends RestApiService {
} catch (Exception e) {
finish = true;
+ t = e;
return Response.ok().entity(ExceptionHandler.tryCatchException(e)).build();
} finally {
long costTime = System.nanoTime() - startTime;
StatementType statementType =
Optional.ofNullable(statement)
- .map(s -> s.getType())
+ .map(Statement::getType)
.orElse(StatementType.FAST_LAST_QUERY);
CommonUtils.addStatementExecutionLatency(
@@ -219,10 +226,30 @@ public class RestApiServiceImpl extends RestApiService {
}
if (queryId != null) {
COORDINATOR.cleanupQueryExecution(queryId);
+ } else {
+ recordQueries(() -> costTime, new FastLastQueryContentSupplier(prefixPathList), t);
}
}
}
+ private static class FastLastQueryContentSupplier implements Supplier<String> {
+
+ private final PrefixPathList prefixPath;
+
+ private FastLastQueryContentSupplier(PrefixPathList prefixPath) {
+ this.prefixPath = prefixPath;
+ }
+
+ @Override
+ public String get() {
+ return restFastLastQuerySql(prefixPath);
+ }
+ }
+
+ public static String restFastLastQuerySql(PrefixPathList prefixPath) {
+ return String.format(FORMAT, String.join(".", prefixPath.getPrefixPaths()));
+ }
+
@Override
public Response executeNonQueryStatement(SQL sql, SecurityContext securityContext) {
Long queryId = null;
@@ -262,7 +289,7 @@ public class RestApiServiceImpl extends RestApiService {
sql.getSql(),
partitionFetcher,
schemaFetcher,
- config.getQueryTimeoutThreshold(),
+ CONFIG.getQueryTimeoutThreshold(),
false);
finish = true;
return responseGenerateHelper(result);
@@ -328,7 +355,7 @@ public class RestApiServiceImpl extends RestApiService {
sql.getSql(),
partitionFetcher,
schemaFetcher,
- config.getQueryTimeoutThreshold(),
+ CONFIG.getQueryTimeoutThreshold(),
true);
finish = true;
if (result.status.code != TSStatusCode.SUCCESS_STATUS.getStatusCode()
@@ -394,7 +421,7 @@ public class RestApiServiceImpl extends RestApiService {
"",
partitionFetcher,
schemaFetcher,
- config.getQueryTimeoutThreshold(),
+ CONFIG.getQueryTimeoutThreshold(),
false);
return responseGenerateHelper(result);
@@ -449,7 +476,7 @@ public class RestApiServiceImpl extends RestApiService {
"",
partitionFetcher,
schemaFetcher,
- config.getQueryTimeoutThreshold(),
+ CONFIG.getQueryTimeoutThreshold(),
false);
return responseGenerateHelper(result);
} catch (Exception e) {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/ClientRPCServiceImpl.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/ClientRPCServiceImpl.java
index 458ec796fd8..256bb53b3d9 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/ClientRPCServiceImpl.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/ClientRPCServiceImpl.java
@@ -223,6 +223,7 @@ import java.util.concurrent.TimeUnit;
import static org.apache.iotdb.commons.partition.DataPartition.NOT_ASSIGNED;
import static org.apache.iotdb.db.queryengine.common.DataNodeEndPoints.isSameNode;
import static org.apache.iotdb.db.queryengine.execution.operator.AggregationUtil.initTimeRangeIterator;
+import static org.apache.iotdb.db.queryengine.plan.Coordinator.recordQueries;
import static org.apache.iotdb.db.utils.CommonUtils.getContentOfRequest;
import static org.apache.iotdb.db.utils.CommonUtils.getContentOfTSFastLastDataQueryForOneDeviceReq;
import static org.apache.iotdb.db.utils.ErrorHandlingUtils.onIoTDBException;
@@ -936,6 +937,7 @@ public class ClientRPCServiceImpl implements IClientRPCServiceWithHandler {
@Override
public TSExecuteStatementResp executeFastLastDataQueryForOnePrefixPath(
final TSFastLastDataQueryForOnePrefixPathReq req) {
+ long startTime = System.nanoTime();
final IClientSession clientSession = SESSION_MANAGER.getCurrSessionAndUpdateIdleTime();
if (!SESSION_MANAGER.checkLogin(clientSession)) {
return RpcUtils.getTSExecuteStatementResp(getNotLoggedInStatus());
@@ -1006,6 +1008,14 @@ public class ClientRPCServiceImpl implements IClientRPCServiceWithHandler {
}
resp.setMoreData(false);
+
+ long costTime = System.nanoTime() - startTime;
+
+ CommonUtils.addStatementExecutionLatency(
+ OperationType.EXECUTE_QUERY_STATEMENT, StatementType.FAST_LAST_QUERY.name(), costTime);
+ CommonUtils.addQueryLatency(StatementType.FAST_LAST_QUERY, costTime);
+ recordQueries(
+ () -> costTime, () -> String.format("thrift fastLastQuery %s", prefixPath), null);
return resp;
} catch (final Exception e) {
return RpcUtils.getTSExecuteStatementResp(
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/common/MPPQueryContext.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/common/MPPQueryContext.java
index 3fca60d5507..9d71f9147ac 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/common/MPPQueryContext.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/common/MPPQueryContext.java
@@ -77,13 +77,13 @@ public class MPPQueryContext {
private boolean isExplainAnalyze = false;
- QueryPlanStatistics queryPlanStatistics = null;
+ private QueryPlanStatistics queryPlanStatistics = null;
// To avoid query front-end from consuming too much memory, it needs to reserve memory when
// constructing some Expression and PlanNode.
private final MemoryReservationManager memoryReservationManager;
- private static final int minSizeToUseSampledTimeseriesOperandMemCost = 100;
+ private static final int MIN_SIZE_TO_USE_SAMPLED_TIMESERIES_OPERAND_MEM_COST = 100;
private double avgTimeseriesOperandMemCost = 0;
private int numsOfSampledTimeseriesOperand = 0;
// When there is no view in a last query and no device exists in multiple regions,
@@ -103,7 +103,6 @@ public class MPPQueryContext {
new NotThreadSafeMemoryReservationManager(queryId, this.getClass().getName());
}
- // TODO too many callers just pass a null SessionInfo which should be forbidden
public MPPQueryContext(
String sql,
QueryId queryId,
@@ -389,7 +388,7 @@ public class MPPQueryContext {
}
public boolean useSampledAvgTimeseriesOperandMemCost() {
- return numsOfSampledTimeseriesOperand >= minSizeToUseSampledTimeseriesOperandMemCost;
+ return numsOfSampledTimeseriesOperand >= MIN_SIZE_TO_USE_SAMPLED_TIMESERIES_OPERAND_MEM_COST;
}
public long getAvgTimeseriesOperandMemCost() {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/Coordinator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/Coordinator.java
index ca72d2f3151..ebb843735e9 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/Coordinator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/Coordinator.java
@@ -122,6 +122,7 @@ import org.apache.iotdb.db.queryengine.plan.statement.IConfigStatement;
import org.apache.iotdb.db.queryengine.plan.statement.Statement;
import org.apache.iotdb.db.utils.SetThreadName;
+import org.apache.thrift.TBase;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -132,6 +133,8 @@ import java.util.concurrent.ExecutorService;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.function.BiFunction;
+import java.util.function.LongSupplier;
+import java.util.function.Supplier;
import static org.apache.iotdb.commons.utils.StatusUtils.needRetry;
import static org.apache.iotdb.db.utils.CommonUtils.getContentOfRequest;
@@ -221,7 +224,7 @@ public class Coordinator {
QueryId globalQueryId = queryIdGenerator.createNextQueryId();
MPPQueryContext queryContext = null;
try (SetThreadName queryName = new SetThreadName(globalQueryId.getId())) {
- if (sql != null && !sql.isEmpty()) {
+ if (LOGGER.isDebugEnabled() && sql != null && !sql.isEmpty()) {
LOGGER.debug("[QueryStart] sql: {}", sql);
}
queryContext =
@@ -532,32 +535,54 @@ public class Coordinator {
queryExecution.stopAndCleanup(t);
queryExecutionMap.remove(queryId);
if (queryExecution.isQuery() && queryExecution.isUserQuery()) {
- long costTime = queryExecution.getTotalExecutionTime();
- // print slow query
- if (costTime / 1_000_000 >= CONFIG.getSlowQueryThreshold()) {
- SLOW_SQL_LOGGER.info(
- "Cost: {} ms, {}",
- costTime / 1_000_000,
- getContentOfRequest(nativeApiRequest, queryExecution));
- }
-
- // only sample successful query
- if (t == null && COMMON_CONFIG.isEnableQuerySampling()) { // sampling is enabled
- String queryRequest = getContentOfRequest(nativeApiRequest, queryExecution);
- if (COMMON_CONFIG.isQuerySamplingHasRateLimit()) {
- if (COMMON_CONFIG.getQuerySamplingRateLimiter().tryAcquire(queryRequest.length())) {
- SAMPLED_QUERIES_LOGGER.info(queryRequest);
- }
- } else {
- // no limit, always sampled
- SAMPLED_QUERIES_LOGGER.info(queryRequest);
- }
- }
+ recordQueries(
+ queryExecution::getTotalExecutionTime,
+ new ContentOfQuerySupplier(nativeApiRequest, queryExecution),
+ t);
}
}
}
}
+ private static class ContentOfQuerySupplier implements Supplier<String> {
+
+ private final org.apache.thrift.TBase<?, ?> nativeApiRequest;
+ private final IQueryExecution queryExecution;
+
+ private ContentOfQuerySupplier(TBase<?, ?> nativeApiRequest, IQueryExecution queryExecution) {
+ this.nativeApiRequest = nativeApiRequest;
+ this.queryExecution = queryExecution;
+ }
+
+ @Override
+ public String get() {
+ return getContentOfRequest(nativeApiRequest, queryExecution);
+ }
+ }
+
+ public static void recordQueries(
+ LongSupplier executionTime, Supplier<String> contentOfQuerySupplier, Throwable t) {
+
+ long costTime = executionTime.getAsLong();
+ // print slow query
+ if (costTime / 1_000_000 >= CONFIG.getSlowQueryThreshold()) {
+ SLOW_SQL_LOGGER.info("Cost: {} ms, {}", costTime / 1_000_000, contentOfQuerySupplier.get());
+ }
+
+ // only sample successful query
+ if (t == null && COMMON_CONFIG.isEnableQuerySampling()) { // sampling is enabled
+ String queryRequest = contentOfQuerySupplier.get();
+ if (COMMON_CONFIG.isQuerySamplingHasRateLimit()) {
+ if (COMMON_CONFIG.getQuerySamplingRateLimiter().tryAcquire(queryRequest.length())) {
+ SAMPLED_QUERIES_LOGGER.info(queryRequest);
+ }
+ } else {
+ // no limit, always sampled
+ SAMPLED_QUERIES_LOGGER.info(queryRequest);
+ }
+ }
+ }
+
public void cleanupQueryExecution(Long queryId) {
cleanupQueryExecution(queryId, null, null);
}