This is an automated email from the ASF dual-hosted git repository.
jackietien pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 8784b31fc8b Support print fast last query request in sample query log
8784b31fc8b is described below
commit 8784b31fc8b797decfcb9dc864850014b5ddfe64
Author: Jackie Tien <[email protected]>
AuthorDate: Fri Aug 22 08:55:23 2025 +0800
Support print fast last query request in sample query log
---
.../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 +++++++++++++++-------
4 files changed, 95 insertions(+), 34 deletions(-)
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..66be144edef 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,
- "",
+ restFastLastQueryReq(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 restFastLastQueryReq(prefixPath);
+ }
+ }
+
+ public static String restFastLastQueryReq(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);
}
