database/doris
2
0
Fork 0
Code Issues Pull Requests Actions 3 Packages Projects Releases Wiki Activity

[Log](query) Add log of fragment instance num for query (#20597)

Browse Source 
Co-authored-by: weizuo <weizuo@xiaomi.com>
This commit is contained in:
weizuo93
2023-06-15 09:52:13 +08:00
committed by GitHub
parent d8a38f8128
commit 71e8cb061c
2 changed files with 28 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
fe/fe-core/src/main/java/org/apache/doris/qe/Coordinator.java
View File

@ -187,6 +187,7 @@ public class Coordinator {
private final Map<PlanFragmentId, FragmentExecParams> fragmentExecParamsMap = Maps.newHashMap();
private final List<PlanFragment> fragments;
private int instanceTotalNum;
private Map<Long, BackendExecStates> beToExecStates = Maps.newHashMap();
private Map<Long, PipelineExecContexts> beToPipelineExecCtxs = Maps.newHashMap();
@ -482,6 +483,10 @@ public class Coordinator {
return result;
}
public int getInstanceTotalNum() {
return instanceTotalNum;
}
// Initialize
private void prepare() {
for (PlanFragment fragment : fragments) {
@ -664,6 +669,7 @@ public class Coordinator {
// 1. set up exec states
int instanceNum = params.instanceExecParams.size();
Preconditions.checkState(instanceNum > 0);
instanceTotalNum += instanceNum;
List<TExecPlanFragmentParams> tParams = params.toThrift(backendIdx);
// 2. update memory limit for colocate join

22
fe/fe-core/src/main/java/org/apache/doris/qe/StmtExecutor.java
View File

@ -1371,6 +1371,18 @@ public class StmtExecutor {
}
profile.getSummaryProfile().setQueryScheduleFinishTime();
updateProfile(false);
if (coord.getInstanceTotalNum() > 1 && LOG.isDebugEnabled()) {
try {
LOG.debug("Start to execute fragment. user: {}, db: {}, sql: {}, fragment instance num: {}",
context.getQualifiedUser(), context.getDatabase(),
parsedStmt.getOrigStmt().originStmt.replace("\n", " "),
coord.getInstanceTotalNum());
} catch (Exception e) {
LOG.warn("Fail to print fragment concurrency for Query.", e);
}
}
Span fetchResultSpan = context.getTracer().spanBuilder("fetch result").setParent(Context.current()).startSpan();
try (Scope scope = fetchResultSpan.makeCurrent()) {
while (true) {
@ -1449,6 +1461,16 @@ public class StmtExecutor {
throw e;
} finally {
fetchResultSpan.end();
if (coord.getInstanceTotalNum() > 1 && LOG.isDebugEnabled()) {
try {
LOG.debug("Finish to execute fragment. user: {}, db: {}, sql: {}, fragment instance num: {}",
context.getQualifiedUser(), context.getDatabase(),
parsedStmt.getOrigStmt().originStmt.replace("\n", " "),
coord.getInstanceTotalNum());
} catch (Exception e) {
LOG.warn("Fail to print fragment concurrency for Query.", e);
}
}
}
}