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 <[email protected]> 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); }
