Merge branch 'master' into load_step2

fe/fe-core/src/main/java/org/apache/doris/job/base/AbstractJob.java

@@ -78,6 +78,9 @@ public abstract class AbstractJob<T extends AbstractTask, C> implements Job<T, C
     @SerializedName(value = "ctms")
     private Long createTimeMs;
 
+    @SerializedName(value = "ftm")
+    private long finishTimeMs;
+
     @SerializedName(value = "sql")
     String executeSql;
 

fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/InsertJob.java

@@ -150,7 +150,7 @@ public class InsertJob extends AbstractJob<InsertTask, Map> {
     private List<TabletCommitInfo> commitInfos = new ArrayList<>();
 
     // max save task num, do we need to config it?
-    private static final int MAX_SAVE_TASK_NUM = 50;
+    private static final int MAX_SAVE_TASK_NUM = 100;
 
     /**
      * load job type
@@ -204,7 +204,7 @@ public InsertJob(ConnectContext ctx,
     }
 
     @Override
-    public List<InsertTask> createTasks(TaskType taskType) {
+    public List<InsertTask> createTasks(TaskType taskType, Map taskContext) {
         if (plans.isEmpty()) {
             InsertTask task = new InsertTask(labelName, getCurrentDbName(), getExecuteSql(), getCreateUser());
             task.setJobId(getJobId());
@@ -233,7 +233,6 @@ private List<InsertTask> createBatchTasks(TaskType taskType) {
     }
 
     public void addNewTask(long id) {
-
         if (CollectionUtils.isEmpty(taskIdList)) {
             taskIdList = new ConcurrentLinkedQueue<>();
             Env.getCurrentEnv().getEditLog().logUpdateJob(this);
@@ -266,6 +265,12 @@ public void cancelJob(FailMsg failMsg) throws DdlException {
         }
     }
 
+    @Override
+    public boolean isReadyForScheduling(Map taskContext) {
+        return CollectionUtils.isEmpty(getRunningTasks());
+    }
+
+
     @Override
     public void cancelAllTasks() throws JobException {
         super.cancelAllTasks();

fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/InsertTask.java

@@ -184,6 +184,10 @@ protected TUniqueId generateQueryId(String taskIdString) {
     @Override
     public void run() throws JobException {
         try {
+            if (isCanceled.get()) {
+                log.info("task has been canceled, task id is {}", getTaskId());
+                return;
+            }
             command.statefulRun(ctx, stmtExecutor);
         } catch (Exception e) {
             log.warn("execute insert task error, job id is {}, task id is {},sql is {}", getJobId(),
@@ -216,49 +220,6 @@ public void cancel() throws JobException {
         super.cancel();
     }
 
-    @Override
-    public List<String> getShowInfo() {
-        if (null == loadJob) {
-            return new ArrayList<>();
-        }
-        List<String> jobInfo = Lists.newArrayList();
-        // jobId
-        jobInfo.add(String.valueOf(loadJob.getId()));
-        // label
-        jobInfo.add(loadJob.getLabel());
-        // state
-        jobInfo.add(loadJob.getState().name());
-
-        // etl info
-        if (loadJob.getLoadingStatus().getCounters().isEmpty()) {
-            jobInfo.add(FeConstants.null_string);
-        } else {
-            jobInfo.add(Joiner.on("; ").withKeyValueSeparator("=").join(loadJob.getLoadingStatus().getCounters()));
-        }
-
-        // task info
-        jobInfo.add("cluster:" + loadJob.getResourceName() + "; timeout(s):" + loadJob.getTimeout()
-                + "; max_filter_ratio:" + loadJob.getMaxFilterRatio() + "; priority:" + loadJob.getPriority());
-        // error msg
-        if (loadJob.getFailMsg() == null) {
-            jobInfo.add(FeConstants.null_string);
-        } else {
-            jobInfo.add("type:" + loadJob.getFailMsg().getCancelType() + "; msg:" + loadJob.getFailMsg().getMsg());
-        }
-
-        // create time
-        jobInfo.add(TimeUtils.longToTimeString(loadJob.getCreateTimestamp()));
-
-        // load end time
-        jobInfo.add(TimeUtils.longToTimeString(loadJob.getFinishTimestamp()));
-        // tracking url
-        jobInfo.add(loadJob.getLoadingStatus().getTrackingUrl());
-        // jobInfo.add(coordinator.getLoadStatistic().toJson());
-        // user
-        jobInfo.add(loadJob.getUserInfo().getQualifiedUser());
-        return jobInfo;
-    }
-
     @Override
     public TRow getTvfInfo() {
         TRow trow = new TRow();