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 3b246f1f551 [Fix](query) Fix the calc logic of dispatch read time
(#15036)
3b246f1f551 is described below
commit 3b246f1f5513c5c6dc30e3d9f6ad089d52144854
Author: Beyyes <[email protected]>
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 6bdc5c20f64..db0c270d176 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
@@ -129,34 +129,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) {
