You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ca...@apache.org on 2025/03/11 11:03:15 UTC
(iotdb) 01/01: [Fix](query) Fix the calc logic of dispatch read time (#15036)
This is an automated email from the ASF dual-hosted git repository.
caogaofei pushed a commit to branch cp_dispatch_read
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit f8057b3405113e1b6f44821d0ae758afe0d2bd7c
Author: Beyyes <cg...@foxmail.com>
AuthorDate: Mon Mar 10 08:49:07 2025 +0800
[Fix](query) Fix the calc logic of dispatch read time (#15036)
---
.../scheduler/FragmentInstanceDispatcherImpl.java | 54 ++++++++++++----------
1 file changed, 29 insertions(+), 25 deletions(-)
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/FragmentInstanceDispatcherImpl.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/FragmentInstanceDispatcherImpl.java
index 7b4a4e30490..3503201e691 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/FragmentInstanceDispatcherImpl.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/FragmentInstanceDispatcherImpl.java
@@ -126,34 +126,38 @@ public class FragmentInstanceDispatcherImpl implements IFragInstanceDispatcher {
// topological dispatch according to dependency relations between FragmentInstances
private Future<FragInstanceDispatchResult> dispatchRead(List<FragmentInstance> instances) {
long startTime = System.nanoTime();
- for (FragmentInstance instance : instances) {
- try (SetThreadName threadName = new SetThreadName(instance.getId().getFullId())) {
- dispatchOneInstance(instance);
- } catch (FragmentInstanceDispatchException e) {
- return immediateFuture(new FragInstanceDispatchResult(e.getFailureStatus()));
- } catch (Throwable t) {
- LOGGER.warn(DISPATCH_FAILED, t);
- return immediateFuture(
- new FragInstanceDispatchResult(
- RpcUtils.getStatus(
- TSStatusCode.INTERNAL_SERVER_ERROR, UNEXPECTED_ERRORS + t.getMessage())));
- } finally {
- // friendly for gc, clear the plan node tree, for some queries select all devices, it will
- // release lots of memory
- if (!queryContext.isExplainAnalyze()) {
- // EXPLAIN ANALYZE will use these instances, so we can't clear them
- instance.getFragment().clearUselessField();
- } else {
- // TypeProvider is not used in EXPLAIN ANALYZE, so we can clear it
- instance.getFragment().clearTypeProvider();
- }
- long dispatchReadTime = System.nanoTime() - startTime;
- QUERY_EXECUTION_METRIC_SET.recordExecutionCost(DISPATCH_READ, dispatchReadTime);
- queryContext.recordDispatchCost(dispatchReadTime);
+ try {
+ for (FragmentInstance instance : instances) {
+ try (SetThreadName threadName = new SetThreadName(instance.getId().getFullId())) {
+ dispatchOneInstance(instance);
+ } catch (FragmentInstanceDispatchException e) {
+ return immediateFuture(new FragInstanceDispatchResult(e.getFailureStatus()));
+ } catch (Throwable t) {
+ LOGGER.warn(DISPATCH_FAILED, t);
+ return immediateFuture(
+ new FragInstanceDispatchResult(
+ RpcUtils.getStatus(
+ TSStatusCode.INTERNAL_SERVER_ERROR, UNEXPECTED_ERRORS + t.getMessage())));
+ } finally {
+ // friendly for gc, clear the plan node tree, for some queries select all devices, it will
+ // release lots of memory
+ if (!queryContext.isExplainAnalyze()) {
+ // EXPLAIN ANALYZE will use these instances, so we can't clear them
+ instance.getFragment().clearUselessField();
+ } else {
+ // TypeProvider is not used in EXPLAIN ANALYZE, so we can clear it
+ instance.getFragment().clearTypeProvider();
+ }
+ }
}
+
+ return immediateFuture(new FragInstanceDispatchResult(true));
+ } finally {
+ long queryDispatchReadTime = System.nanoTime() - startTime;
+ QUERY_EXECUTION_METRIC_SET.recordExecutionCost(DISPATCH_READ, queryDispatchReadTime);
+ queryContext.recordDispatchCost(queryDispatchReadTime);
}
- return immediateFuture(new FragInstanceDispatchResult(true));
}
private Future<FragInstanceDispatchResult> dispatchWriteSync(List<FragmentInstance> instances) {