morningman commented on code in PR #22224:
URL: https://github.com/apache/doris/pull/22224#discussion_r1275125295
##########
fe/fe-core/src/main/java/org/apache/doris/statistics/AnalysisManager.java:
##########
@@ -700,6 +709,29 @@ public List<AnalysisInfo> showAnalysisJob(ShowAnalyzeStmt
stmt) {
.collect(Collectors.toList());
}
+ public String getJobProgress(long jobId) {
+ List<AnalysisInfo> tasks = findTasks(jobId);
+ int finished = 0;
+ int failed = 0;
+ int inProgress = 0;
+ int total = tasks.size();
+ for (AnalysisInfo info : tasks) {
+ switch (info.state) {
+ case FINISHED:
+ finished++;
+ break;
+ case FAILED:
+ failed++;
+ break;
+ default:
+ inProgress++;
+ break;
+ }
+ }
+ return String.format("%d/%d finished, %d/%d failed, %d/%d in progress",
Review Comment:
How about: `finished/failed/in progress/total` ?
##########
fe/fe-core/src/main/java/org/apache/doris/statistics/AnalysisManager.java:
##########
@@ -813,21 +845,26 @@ public void cancel() {
public void execute() {
List<String> colNames = new ArrayList<>();
+ List<String> errorMessages = new ArrayList<>();
for (BaseAnalysisTask task : tasks) {
if (cancelled) {
+ colNames.add(task.info.colName);
+ errorMessages.add("Cancelled");
continue;
}
try {
task.doExecute();
updateSyncTaskStatus(task, AnalysisState.FINISHED);
} catch (Throwable t) {
colNames.add(task.info.colName);
+ errorMessages.add(t.getMessage());
Review Comment:
Util.getRootCauseMessage(t)
##########
fe/fe-core/src/main/java/org/apache/doris/statistics/AnalysisManager.java:
##########
@@ -578,63 +580,70 @@ private void createTaskForExternalTable(AnalysisInfo
jobInfo,
}
AnalysisInfoBuilder colTaskInfoBuilder = new
AnalysisInfoBuilder(jobInfo);
long taskId = Env.getCurrentEnv().getNextId();
- AnalysisInfo analysisInfo = colTaskInfoBuilder.setIndexId(-1L)
- .setTaskId(taskId).setExternalTableLevelTask(true).build();
+ AnalysisInfo analysisInfo =
colTaskInfoBuilder.setIndexId(-1L).setLastExecTimeInMs(System.currentTimeMillis())
+
.setTaskId(taskId).setColName("TableRowCount").setExternalTableLevelTask(true).build();
analysisTasks.put(taskId, createTask(analysisInfo));
if (isSync) {
// For sync job, don't need to persist, return here and execute it
immediately.
return;
}
try {
- logCreateAnalysisJob(analysisInfo);
+ logCreateAnalysisTask(analysisInfo);
} catch (Exception e) {
throw new DdlException("Failed to create analysis task", e);
}
}
- public void updateTaskStatus(AnalysisInfo info, AnalysisState jobState,
String message, long time) {
+ public void updateTaskStatus(AnalysisInfo info, AnalysisState taskState,
String message, long time) {
if (analysisJobIdToTaskMap.get(info.jobId) == null) {
return;
}
- info.state = jobState;
+ info.state = taskState;
info.message = message;
+ if (taskState.equals(AnalysisState.FINISHED) ||
taskState.equals(AnalysisState.FAILED)) {
+ info.timeCostInMs = time - info.lastExecTimeInMs;
+ }
info.lastExecTimeInMs = time;
logCreateAnalysisTask(info);
AnalysisInfo job = analysisJobInfoMap.get(info.jobId);
- job.lastExecTimeInMs = time;
- if (info.state.equals(AnalysisState.RUNNING) &&
!job.state.equals(AnalysisState.PENDING)) {
- job.state = AnalysisState.RUNNING;
- Env.getCurrentEnv().getEditLog().logCreateAnalysisTasks(job);
- }
- boolean allFinished = true;
- boolean hasFailure = false;
- for (BaseAnalysisTask task :
analysisJobIdToTaskMap.get(info.jobId).values()) {
- AnalysisInfo taskInfo = task.info;
- if (taskInfo.state.equals(AnalysisState.RUNNING) ||
taskInfo.state.equals(AnalysisState.PENDING)) {
- allFinished = false;
- break;
+ // Synchronize the job state change in job level.
+ synchronized (job) {
+ job.lastExecTimeInMs = time;
+ // Set the job state to RUNNING when its first task becomes
RUNNING.
+ if (info.state.equals(AnalysisState.RUNNING) &&
job.state.equals(AnalysisState.PENDING)) {
+ job.state = AnalysisState.RUNNING;
+ logCreateAnalysisJob(job);
}
- if (taskInfo.state.equals(AnalysisState.FAILED)) {
- hasFailure = true;
+ boolean allFinished = true;
+ boolean hasFailure = false;
+ for (BaseAnalysisTask task :
analysisJobIdToTaskMap.get(info.jobId).values()) {
+ AnalysisInfo taskInfo = task.info;
+ if (taskInfo.state.equals(AnalysisState.RUNNING) ||
taskInfo.state.equals(AnalysisState.PENDING)) {
+ allFinished = false;
+ break;
+ }
+ if (taskInfo.state.equals(AnalysisState.FAILED)) {
+ hasFailure = true;
+ }
}
- }
- if (allFinished) {
- if (hasFailure) {
- job.state = AnalysisState.FAILED;
- logCreateAnalysisJob(job);
- } else {
- job.state = AnalysisState.FINISHED;
- if (job.jobType.equals(JobType.SYSTEM)) {
- try {
- updateTableStats(job);
- } catch (Throwable e) {
- LOG.warn("Failed to update Table statistics in job:
{}", info.toString());
+ if (allFinished) {
+ if (hasFailure) {
+ job.state = AnalysisState.FAILED;
+ logCreateAnalysisJob(job);
+ } else {
+ job.state = AnalysisState.FINISHED;
+ if (job.jobType.equals(JobType.SYSTEM)) {
+ try {
+ updateTableStats(job);
+ } catch (Throwable e) {
+ LOG.warn("Failed to update Table statistics in
job: {}", info.toString());
Review Comment:
print `e`
##########
fe/fe-core/src/main/java/org/apache/doris/qe/StmtExecutor.java:
##########
@@ -2481,7 +2482,7 @@ public List<ResultRow> executeInternalQuery() {
analyze(context.getSessionVariable().toThrift());
}
} catch (Exception e) {
- throw new RuntimeException("Failed to execute internal SQL",
e);
+ throw new RuntimeException("Failed to execute internal SQL. "
+ e.getMessage(), e);
Review Comment:
```suggestion
throw new RuntimeException("Failed to execute internal SQL.
" + Util.getRootCauseMessage(e), e);
```
##########
fe/fe-core/src/main/java/org/apache/doris/qe/StmtExecutor.java:
##########
@@ -2481,7 +2482,7 @@ public List<ResultRow> executeInternalQuery() {
analyze(context.getSessionVariable().toThrift());
}
} catch (Exception e) {
- throw new RuntimeException("Failed to execute internal SQL",
e);
+ throw new RuntimeException("Failed to execute internal SQL. "
+ e.getMessage(), e);
Review Comment:
Same for other places
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscr...@doris.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@doris.apache.org
For additional commands, e-mail: commits-h...@doris.apache.org
