You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@doris.apache.org by "CalvinKirs (via GitHub)" <gi...@apache.org> on 2023/11/12 14:55:22 UTC
[PR] [Improve](Job)Refactor JOB [doris]
CalvinKirs opened a new pull request, #26845:
URL: https://github.com/apache/doris/pull/26845
### Abstract
refactor JOB design to support non-timer-scheduled task type jobs and provide a unified job registration interface
### NOTICE
This will cause the master's metadata to be incompatible
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [Improve](Job)Refactor JOB [doris]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #26845:
URL: https://github.com/apache/doris/pull/26845#issuecomment-1809827651
PR approved by anyone and no changes requested.
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [Improve](Job)Refactor JOB [doris]
Posted by "doris-robot (via GitHub)" <gi...@apache.org>.
doris-robot commented on PR #26845:
URL: https://github.com/apache/doris/pull/26845#issuecomment-1809784437
(From new machine)TeamCity pipeline, clickbench performance test result:
the sum of best hot time: 45.75 seconds
stream load tsv: 550 seconds loaded 74807831229 Bytes, about 129 MB/s
stream load json: 18 seconds loaded 2358488459 Bytes, about 124 MB/s
stream load orc: 65 seconds loaded 1101869774 Bytes, about 16 MB/s
stream load parquet: 32 seconds loaded 861443392 Bytes, about 25 MB/s
insert into select: 27.9 seconds inserted 10000000 Rows, about 358K ops/s
storage size: 17096679592 Bytes
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [Improve](Job)Refactor JOB [doris]
Posted by "doris-robot (via GitHub)" <gi...@apache.org>.
doris-robot commented on PR #26845:
URL: https://github.com/apache/doris/pull/26845#issuecomment-1809935795
<details>
<summary>TPC-H test result on machine: 'aliyun_ecs.c7a.8xlarge_32C64G'</summary>
```
Tpch sf100 test result on commit adca62d0575f48c70b563231627937adac7a5b22, data reload: true
run tpch-sf100 query with default conf and session variables
q1 5298 5058 5048 5048
q2 388 127 118 118
q3 2127 2005 2014 2005
q4 1445 1365 1363 1363
q5 4008 3994 3939 3939
q6 254 127 120 120
q7 1415 863 884 863
q8 2823 2771 2759 2759
q9 11292 9681 9580 9580
q10 9935 3535 3516 3516
q11 393 242 254 242
q12 450 279 277 277
q13 4557 4158 4118 4118
q14 305 283 284 283
q15 619 536 553 536
q16 669 586 582 582
q17 1139 1061 1070 1061
q18 8074 7511 7723 7511
q19 1797 1663 1676 1663
q20 528 310 288 288
q21 4669 4342 4372 4342
q22 498 409 404 404
Total cold run time: 62683 ms
Total hot run time: 50618 ms
run tpch-sf100 query with default conf and set session variable runtime_filter_mode=off
q1 5053 5007 4974 4974
q2 325 232 225 225
q3 4047 4011 4051 4011
q4 2761 2766 2731 2731
q5 9745 9721 9785 9721
q6 240 121 115 115
q7 2701 2279 2291 2279
q8 4879 4853 4847 4847
q9 13332 13183 13592 13183
q10 4095 4198 4196 4196
q11 841 655 653 653
q12 981 810 823 810
q13 4299 3890 3917 3890
q14 366 356 336 336
q15 577 566 544 544
q16 733 713 655 655
q17 3878 3952 3894 3894
q18 9614 9321 9331 9321
q19 1845 1783 1773 1773
q20 2424 2053 2083 2053
q21 8929 8722 8945 8722
q22 956 841 858 841
Total cold run time: 82621 ms
Total hot run time: 79774 ms
```
</details>
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [Improve](Job)Refactor JOB [doris]
Posted by "doris-robot (via GitHub)" <gi...@apache.org>.
doris-robot commented on PR #26845:
URL: https://github.com/apache/doris/pull/26845#issuecomment-1809910598
(From new machine)TeamCity pipeline, clickbench performance test result:
the sum of best hot time: 44.67 seconds
stream load tsv: 576 seconds loaded 74807831229 Bytes, about 123 MB/s
stream load json: 18 seconds loaded 2358488459 Bytes, about 124 MB/s
stream load orc: 65 seconds loaded 1101869774 Bytes, about 16 MB/s
stream load parquet: 32 seconds loaded 861443392 Bytes, about 25 MB/s
insert into select: 28.6 seconds inserted 10000000 Rows, about 349K ops/s
storage size: 17096979509 Bytes
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [Improve](Job)Refactor JOB [doris]
Posted by "morningman (via GitHub)" <gi...@apache.org>.
morningman commented on code in PR #26845:
URL: https://github.com/apache/doris/pull/26845#discussion_r1391966588
##########
fe/fe-core/src/main/java/org/apache/doris/job/manager/TaskDisruptorGroupManager.java:
##########
@@ -0,0 +1,101 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.job.manager;
+
+import org.apache.doris.common.CustomThreadFactory;
+import org.apache.doris.job.base.AbstractJob;
+import org.apache.doris.job.base.JobExecutionConfiguration;
+import org.apache.doris.job.common.JobType;
+import org.apache.doris.job.disruptor.ExecuteTaskEvent;
+import org.apache.doris.job.disruptor.TaskDisruptor;
+import org.apache.doris.job.disruptor.TimerJobEvent;
+import org.apache.doris.job.executor.DefaultTaskExecutorHandler;
+import org.apache.doris.job.executor.DispatchTaskHandler;
+import org.apache.doris.job.extensions.insert.InsertTask;
+import org.apache.doris.job.task.AbstractTask;
+
+import com.lmax.disruptor.BlockingWaitStrategy;
+import com.lmax.disruptor.EventFactory;
+import com.lmax.disruptor.EventTranslatorVararg;
+import com.lmax.disruptor.WorkHandler;
+import lombok.Getter;
+
+import java.util.EnumMap;
+import java.util.Map;
+import java.util.concurrent.ThreadFactory;
+
+public class TaskDisruptorGroupManager<T extends AbstractTask> {
Review Comment:
expose register interface
##########
fe/fe-core/src/main/java/org/apache/doris/job/common/TaskType.java:
##########
@@ -0,0 +1,24 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.job.common;
+
+public enum TaskType {
+
+ SCHEDULER,
Review Comment:
name?
##########
fe/fe-core/src/main/java/org/apache/doris/job/executor/DefaultTaskExecutorHandler.java:
##########
@@ -0,0 +1,75 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.job.executor;
+
+import org.apache.doris.job.disruptor.ExecuteTaskEvent;
+import org.apache.doris.job.manager.TaskTokenManager;
+import org.apache.doris.job.task.AbstractTask;
+
+import com.lmax.disruptor.WorkHandler;
+import lombok.extern.slf4j.Slf4j;
+
+import java.util.concurrent.Semaphore;
+
+/**
+ * DefaultTaskExecutor is an implementation of the TaskExecutor interface.
+ * if you need to implement your own TaskExecutor, you could refer to this class. and need to register
+ * it in the TaskExecutorFactory
+ * It executes a given AbstractTask by acquiring a semaphore token from the TaskTokenManager
+ * and releasing it after the task execution.
+ */
+@Slf4j
+public class DefaultTaskExecutorHandler<T extends AbstractTask> implements WorkHandler<ExecuteTaskEvent<T>> {
+
+
+ @Override
+ public void onEvent(ExecuteTaskEvent<T> executeTaskEvent) {
+ T task = executeTaskEvent.getTask();
+ if (null == task) {
+ log.warn("task is null, ignore,maybe task has been canceled");
+ return;
+ }
+ if (task.isCancelled()) {
+ log.info("task is canceled, ignore");
+ return;
+ }
+ if (null == executeTaskEvent.getJobConfig().getMaxConcurrentTaskNum()
+ || executeTaskEvent.getJobConfig().getMaxConcurrentTaskNum() <= 0) {
+ try {
+ task.runTask();
+ return;
+ } catch (Exception e) {
+ log.warn("execute task error, task id is {}", task.getTaskId(), e);
+ }
+ }
+ Semaphore semaphore = null;
Review Comment:
add a TODO here
##########
fe/fe-core/src/main/java/org/apache/doris/job/scheduler/JobScheduler.java:
##########
@@ -0,0 +1,173 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.job.scheduler;
+
+import org.apache.doris.common.CustomThreadFactory;
+import org.apache.doris.common.util.TimeUtils;
+import org.apache.doris.job.base.AbstractJob;
+import org.apache.doris.job.base.JobExecuteType;
+import org.apache.doris.job.common.JobStatus;
+import org.apache.doris.job.common.TaskType;
+import org.apache.doris.job.disruptor.TaskDisruptor;
+import org.apache.doris.job.executor.TimerJobSchedulerTask;
+import org.apache.doris.job.manager.TaskDisruptorGroupManager;
+import org.apache.doris.job.task.AbstractTask;
+
+import io.netty.util.HashedWheelTimer;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.collections.CollectionUtils;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+@Slf4j
+public class JobScheduler<T extends AbstractJob<?>> implements Closeable {
+
+ /**
+ * scheduler tasks, it's used to scheduler job
+ */
+ private HashedWheelTimer timerTaskScheduler;
+
+ private TaskDisruptor timerJobDisruptor;
+
+ private TaskDisruptorGroupManager taskDisruptorGroupManager;
+
+ private long latestBatchSchedulerTimerTaskTimeMs = 0L;
+
+ private static final long BATCH_SCHEDULER_INTERVAL_SECONDS = 60;
+
+ private static final int HASHED_WHEEL_TIMER_TICKS_PER_WHEEL = 660;
+
+ private final Map<Long, T> jobMap;
+
+ public JobScheduler(Map<Long, T> jobMap) {
+ this.jobMap = jobMap;
+ }
+
+ /**
+ * batch scheduler interval ms time
+ */
+ private static final long BATCH_SCHEDULER_INTERVAL_MILLI_SECONDS = BATCH_SCHEDULER_INTERVAL_SECONDS * 1000L;
+
+ public void start() {
+ timerTaskScheduler = new HashedWheelTimer(new CustomThreadFactory("timer-task-scheduler"), 1,
+ TimeUnit.SECONDS, HASHED_WHEEL_TIMER_TICKS_PER_WHEEL);
+ timerTaskScheduler.start();
+ taskDisruptorGroupManager = new TaskDisruptorGroupManager();
+ taskDisruptorGroupManager.init();
+ this.timerJobDisruptor = taskDisruptorGroupManager.getDispatchDisruptor();
+ latestBatchSchedulerTimerTaskTimeMs = System.currentTimeMillis();
+ batchSchedulerTimerJob();
+ cycleSystemSchedulerTasks();
+ }
+
+ /**
+ * We will cycle system scheduler tasks every 10 minutes.
+ * Jobs will be re-registered after the task is completed
+ */
+ private void cycleSystemSchedulerTasks() {
+ log.info("re-register system scheduler timer tasks" + TimeUtils.longToTimeString(System.currentTimeMillis()));
+ timerTaskScheduler.newTimeout(timeout -> {
+ batchSchedulerTimerJob();
+ cycleSystemSchedulerTasks();
+ }, BATCH_SCHEDULER_INTERVAL_SECONDS, TimeUnit.SECONDS);
+
+ }
+
+ private void batchSchedulerTimerJob() {
+ executeTimerJobIdsWithinLastTenMinutesWindow();
+ }
+
+ public void scheduleOneJob(T job) {
+ if (!job.getJobStatus().equals(JobStatus.RUNNING)) {
+ return;
+ }
+ if (!job.getJobConfig().checkIsTimerJob()) {
+ //manual job will not scheduler
+ if (JobExecuteType.MANUAL.equals(job.getJobConfig().getExecuteType())) {
+ return;
+ }
+ //todo skip streaming job,improve in the future
+ if (JobExecuteType.INSTANT.equals(job.getJobConfig().getExecuteType()) && job.isReadyForScheduling()) {
+ schedulerImmediateJob(job);
+ }
+ }
+ //if it's timer job and trigger last window already start, we will scheduler it immediately
+ cycleTimerJobScheduler(job);
+ }
+
+ @Override
+ public void close() throws IOException {
+
+ }
+
+
+ private void cycleTimerJobScheduler(T job) {
+ List<Long> delaySeconds = job.getJobConfig().getTriggerDelayTimes(System.currentTimeMillis(),
+ System.currentTimeMillis(), latestBatchSchedulerTimerTaskTimeMs);
+ if (CollectionUtils.isNotEmpty(delaySeconds)) {
+ delaySeconds.forEach(delaySecond -> {
+ TimerJobSchedulerTask<T> timerJobSchedulerTask = new TimerJobSchedulerTask<>(timerJobDisruptor, job);
+ timerTaskScheduler.newTimeout(timerJobSchedulerTask, delaySecond, TimeUnit.SECONDS);
+ });
+ }
+ }
+
+
+ private void schedulerImmediateJob(T job) {
Review Comment:
name unify
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [WIP][Improve](Job)Refactor JOB [doris]
Posted by "CalvinKirs (via GitHub)" <gi...@apache.org>.
CalvinKirs commented on code in PR #26845:
URL: https://github.com/apache/doris/pull/26845#discussion_r1391869195
##########
fe/fe-core/src/main/java/org/apache/doris/job/scheduler/JobScheduler.java:
##########
@@ -0,0 +1,171 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.job.scheduler;
+
+import org.apache.doris.common.CustomThreadFactory;
+import org.apache.doris.common.util.TimeUtils;
+import org.apache.doris.job.base.AbstractJob;
+import org.apache.doris.job.base.JobExecuteType;
+import org.apache.doris.job.common.JobStatus;
+import org.apache.doris.job.common.TaskType;
+import org.apache.doris.job.disruptor.TaskDisruptor;
+import org.apache.doris.job.executor.TimerJobSchedulerTask;
+import org.apache.doris.job.manager.TaskDisruptorGroupManager;
+import org.apache.doris.job.task.AbstractTask;
+
+import io.netty.util.HashedWheelTimer;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.collections.CollectionUtils;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+@Slf4j
+public class JobScheduler<T extends AbstractJob<?>> implements Closeable {
Review Comment:
We hope to provide a unified job scheduler interface. For example, `MTMV jobs` and `Insert jobs` can be registered and managed through the unified interface.
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [WIP][Improve](Job)Refactor JOB [doris]
Posted by "CalvinKirs (via GitHub)" <gi...@apache.org>.
CalvinKirs commented on code in PR #26845:
URL: https://github.com/apache/doris/pull/26845#discussion_r1391886383
##########
fe/fe-core/src/main/java/org/apache/doris/job/manager/JobManager.java:
##########
@@ -0,0 +1,213 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.job.manager;
+
+import org.apache.doris.catalog.Env;
+import org.apache.doris.common.io.Writable;
+import org.apache.doris.common.util.LogBuilder;
+import org.apache.doris.common.util.LogKey;
+import org.apache.doris.job.base.AbstractJob;
+import org.apache.doris.job.common.JobStatus;
+import org.apache.doris.job.common.JobType;
+import org.apache.doris.job.exception.JobException;
+import org.apache.doris.job.scheduler.JobScheduler;
+import org.apache.doris.job.task.AbstractTask;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+import java.util.List;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.stream.Collectors;
+
+@Slf4j
+public class JobManager<T extends AbstractJob<?>> implements Writable {
Review Comment:
We hope to provide a unified job manager interface. For example, MTMV jobs and Insert jobs can be registered and managed through the unified interface.
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [WIP][Improve](Job)Refactor JOB [doris]
Posted by "doris-robot (via GitHub)" <gi...@apache.org>.
doris-robot commented on PR #26845:
URL: https://github.com/apache/doris/pull/26845#issuecomment-1808125390
<details>
<summary>TPC-H test result on machine: 'aliyun_ecs.c7a.8xlarge_32C64G'</summary>
```
Tpch sf100 test result on commit d10cce6038a367a3d75d22e802fd802b7b3c59b9, data reload: true
run tpch-sf100 query with default conf and session variables
q1 5314 5106 5203 5106
q2 377 138 127 127
q3 2125 2035 2001 2001
q4 3098 1297 1333 1297
q5 4166 3982 3954 3954
q6 255 121 124 121
q7 1430 878 881 878
q8 4771 2782 2767 2767
q9 9837 9679 9565 9565
q10 9933 3554 3517 3517
q11 377 239 256 239
q12 461 283 278 278
q13 4587 4125 4122 4122
q14 311 297 284 284
q15 651 580 577 577
q16 674 579 593 579
q17 1146 1104 1095 1095
q18 8046 7622 7702 7622
q19 1726 1702 1693 1693
q20 542 296 293 293
q21 4705 4381 4377 4377
q22 503 414 419 414
Total cold run time: 65035 ms
Total hot run time: 50906 ms
run tpch-sf100 query with default conf and set session variable runtime_filter_mode=off
q1 5102 5154 5191 5154
q2 323 218 218 218
q3 4121 4070 3989 3989
q4 2807 2761 2750 2750
q5 9774 9832 9782 9782
q6 244 121 118 118
q7 2749 2303 2316 2303
q8 4907 4899 4881 4881
q9 13638 13199 13099 13099
q10 4070 4168 4155 4155
q11 809 659 697 659
q12 1008 828 818 818
q13 4293 3887 3895 3887
q14 385 347 335 335
q15 632 535 569 535
q16 773 701 704 701
q17 3948 3894 3907 3894
q18 9662 9522 9330 9330
q19 1959 1806 1776 1776
q20 2393 2070 2078 2070
q21 8742 8848 8951 8848
q22 950 874 865 865
Total cold run time: 83289 ms
Total hot run time: 80167 ms
```
</details>
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [Improve](Job)Refactor JOB [doris]
Posted by "morningman (via GitHub)" <gi...@apache.org>.
morningman commented on code in PR #26845:
URL: https://github.com/apache/doris/pull/26845#discussion_r1392090982
##########
fe/fe-common/src/main/java/org/apache/doris/common/Config.java:
##########
@@ -1570,10 +1570,46 @@ public class Config extends ConfigBase {
@ConfField
public static boolean enable_pipeline_load = false;
+ /*---------------------- JOB CONFIG START------------------------*/
+ /**
+ * The number of threads used to dispatch timer job.
+ * If we have a lot of timer jobs, we need more threads to dispatch them.
+ * All timer job will be dispatched to a thread pool, and they will be dispatched to the thread queue of the
+ * corresponding type of job
+ * The value should be greater than 0, if it is 0 or <=0, set it to 5
+ */
+ @ConfField
Review Comment:
use `description` field in `@ConfField`
##########
fe/fe-common/src/main/java/org/apache/doris/common/Config.java:
##########
@@ -1570,10 +1570,46 @@ public class Config extends ConfigBase {
@ConfField
public static boolean enable_pipeline_load = false;
+ /*---------------------- JOB CONFIG START------------------------*/
+ /**
+ * The number of threads used to dispatch timer job.
+ * If we have a lot of timer jobs, we need more threads to dispatch them.
+ * All timer job will be dispatched to a thread pool, and they will be dispatched to the thread queue of the
+ * corresponding type of job
+ * The value should be greater than 0, if it is 0 or <=0, set it to 5
+ */
+ @ConfField
+ public static int job_dispatch_timer_job_thread_num = 5;
+
+ /**
+ * The number of timer jobs that can be queued.
+ * if consumer is slow, the queue will be full, and the producer will be blocked.
+ * if you have a lot of timer jobs, you need to increase this value or increase the number of
+ * {@code @dispatch_timer_job_thread_num}
+ * The value should be greater than 0, if it is 0 or <=0, set it to 1024
+ */
+ @ConfField
+ public static int job_dispatch_timer_job_queue_size = 1024;
+ /**
+ * The number of threads used to consume insert tasks.
+ * if you have a lot of insert jobs,and the average execution frequency is relatively high you need to increase
+ * this value or increase the number of {@code @job_insert_task_queue_size}
+ * The value should be greater than 0, if it is 0 or <=0, set it to 5
+ */
@ConfField
- public static int scheduler_job_task_max_saved_count = 20;
+ public static int job_insert_task_consumer_thread_num = 5;
Review Comment:
set to 10, same as previous load loading task
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [Improve](Job)Refactor JOB [doris]
Posted by "CalvinKirs (via GitHub)" <gi...@apache.org>.
CalvinKirs commented on code in PR #26845:
URL: https://github.com/apache/doris/pull/26845#discussion_r1392145017
##########
fe/fe-core/src/main/java/org/apache/doris/job/scheduler/JobScheduler.java:
##########
@@ -0,0 +1,173 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.job.scheduler;
+
+import org.apache.doris.common.CustomThreadFactory;
+import org.apache.doris.common.util.TimeUtils;
+import org.apache.doris.job.base.AbstractJob;
+import org.apache.doris.job.base.JobExecuteType;
+import org.apache.doris.job.common.JobStatus;
+import org.apache.doris.job.common.TaskType;
+import org.apache.doris.job.disruptor.TaskDisruptor;
+import org.apache.doris.job.executor.TimerJobSchedulerTask;
+import org.apache.doris.job.manager.TaskDisruptorGroupManager;
+import org.apache.doris.job.task.AbstractTask;
+
+import io.netty.util.HashedWheelTimer;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.collections.CollectionUtils;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+@Slf4j
+public class JobScheduler<T extends AbstractJob<?>> implements Closeable {
+
+ /**
+ * scheduler tasks, it's used to scheduler job
+ */
+ private HashedWheelTimer timerTaskScheduler;
+
+ private TaskDisruptor timerJobDisruptor;
+
+ private TaskDisruptorGroupManager taskDisruptorGroupManager;
+
+ private long latestBatchSchedulerTimerTaskTimeMs = 0L;
+
+ private static final long BATCH_SCHEDULER_INTERVAL_SECONDS = 60;
+
+ private static final int HASHED_WHEEL_TIMER_TICKS_PER_WHEEL = 660;
+
+ private final Map<Long, T> jobMap;
+
+ public JobScheduler(Map<Long, T> jobMap) {
+ this.jobMap = jobMap;
+ }
+
+ /**
+ * batch scheduler interval ms time
+ */
+ private static final long BATCH_SCHEDULER_INTERVAL_MILLI_SECONDS = BATCH_SCHEDULER_INTERVAL_SECONDS * 1000L;
+
+ public void start() {
+ timerTaskScheduler = new HashedWheelTimer(new CustomThreadFactory("timer-task-scheduler"), 1,
+ TimeUnit.SECONDS, HASHED_WHEEL_TIMER_TICKS_PER_WHEEL);
+ timerTaskScheduler.start();
+ taskDisruptorGroupManager = new TaskDisruptorGroupManager();
+ taskDisruptorGroupManager.init();
+ this.timerJobDisruptor = taskDisruptorGroupManager.getDispatchDisruptor();
+ latestBatchSchedulerTimerTaskTimeMs = System.currentTimeMillis();
+ batchSchedulerTimerJob();
+ cycleSystemSchedulerTasks();
+ }
+
+ /**
+ * We will cycle system scheduler tasks every 10 minutes.
+ * Jobs will be re-registered after the task is completed
+ */
+ private void cycleSystemSchedulerTasks() {
+ log.info("re-register system scheduler timer tasks" + TimeUtils.longToTimeString(System.currentTimeMillis()));
+ timerTaskScheduler.newTimeout(timeout -> {
+ batchSchedulerTimerJob();
+ cycleSystemSchedulerTasks();
+ }, BATCH_SCHEDULER_INTERVAL_SECONDS, TimeUnit.SECONDS);
+
+ }
+
+ private void batchSchedulerTimerJob() {
+ executeTimerJobIdsWithinLastTenMinutesWindow();
+ }
+
+ public void scheduleOneJob(T job) {
+ if (!job.getJobStatus().equals(JobStatus.RUNNING)) {
+ return;
+ }
+ if (!job.getJobConfig().checkIsTimerJob()) {
+ //manual job will not scheduler
+ if (JobExecuteType.MANUAL.equals(job.getJobConfig().getExecuteType())) {
+ return;
+ }
+ //todo skip streaming job,improve in the future
+ if (JobExecuteType.INSTANT.equals(job.getJobConfig().getExecuteType()) && job.isReadyForScheduling()) {
+ schedulerImmediateJob(job);
+ }
+ }
+ //if it's timer job and trigger last window already start, we will scheduler it immediately
+ cycleTimerJobScheduler(job);
+ }
+
+ @Override
+ public void close() throws IOException {
+
+ }
+
+
+ private void cycleTimerJobScheduler(T job) {
+ List<Long> delaySeconds = job.getJobConfig().getTriggerDelayTimes(System.currentTimeMillis(),
+ System.currentTimeMillis(), latestBatchSchedulerTimerTaskTimeMs);
+ if (CollectionUtils.isNotEmpty(delaySeconds)) {
+ delaySeconds.forEach(delaySecond -> {
+ TimerJobSchedulerTask<T> timerJobSchedulerTask = new TimerJobSchedulerTask<>(timerJobDisruptor, job);
+ timerTaskScheduler.newTimeout(timerJobSchedulerTask, delaySecond, TimeUnit.SECONDS);
+ });
+ }
+ }
+
+
+ private void schedulerImmediateJob(T job) {
+ List<? extends AbstractTask> tasks = job.createTasks(TaskType.MANUAL);
+ if (CollectionUtils.isEmpty(tasks)) {
+ return;
Review Comment:
oops...fixed
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [WIP][Improve](Job)Refactor JOB [doris]
Posted by "doris-robot (via GitHub)" <gi...@apache.org>.
doris-robot commented on PR #26845:
URL: https://github.com/apache/doris/pull/26845#issuecomment-1808745195
(From new machine)TeamCity pipeline, clickbench performance test result:
the sum of best hot time: 46.06 seconds
stream load tsv: 554 seconds loaded 74807831229 Bytes, about 128 MB/s
stream load json: 20 seconds loaded 2358488459 Bytes, about 112 MB/s
stream load orc: 65 seconds loaded 1101869774 Bytes, about 16 MB/s
stream load parquet: 32 seconds loaded 861443392 Bytes, about 25 MB/s
insert into select: 29.3 seconds inserted 10000000 Rows, about 341K ops/s
storage size: 17162176768 Bytes
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [Improve](Job)Refactor JOB [doris]
Posted by "CalvinKirs (via GitHub)" <gi...@apache.org>.
CalvinKirs commented on code in PR #26845:
URL: https://github.com/apache/doris/pull/26845#discussion_r1392143170
##########
fe/fe-core/src/main/java/org/apache/doris/analysis/CreateJobStmt.java:
##########
@@ -81,16 +81,17 @@ public class CreateJobStmt extends DdlStmt {
private final String endsTimeStamp;
private final String comment;
+ private JobExecuteType executeType;
private String timezone = TimeUtils.DEFAULT_TIME_ZONE;
private static final ImmutableSet<Class<? extends DdlStmt>> supportStmtSuperClass
= new ImmutableSet.Builder<Class<? extends DdlStmt>>().add(InsertStmt.class)
Review Comment:
This will be implemented together in insertjob
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [WIP][Improve](Job)Refactor JOB [doris]
Posted by "doris-robot (via GitHub)" <gi...@apache.org>.
doris-robot commented on PR #26845:
URL: https://github.com/apache/doris/pull/26845#issuecomment-1808775041
<details>
<summary>TPC-H test result on machine: 'aliyun_ecs.c7a.8xlarge_32C64G'</summary>
```
Tpch sf100 test result on commit 40cd59a3c2f197494520f86b6c166824aa9c95dd, data reload: true
run tpch-sf100 query with default conf and session variables
q1 5378 5213 5087 5087
q2 413 126 125 125
q3 2067 2043 1986 1986
q4 1410 1343 1345 1343
q5 4015 3974 3939 3939
q6 251 123 121 121
q7 1453 882 893 882
q8 2775 2767 2742 2742
q9 11833 9524 9411 9411
q10 9928 3535 3549 3535
q11 403 237 254 237
q12 451 279 289 279
q13 4701 4112 4159 4112
q14 309 280 283 280
q15 629 520 551 520
q16 666 591 576 576
q17 1129 1067 1081 1067
q18 8133 7677 7760 7677
q19 1686 1682 1668 1668
q20 573 326 285 285
q21 4678 4357 4387 4357
q22 525 410 411 410
Total cold run time: 63406 ms
Total hot run time: 50639 ms
run tpch-sf100 query with default conf and set session variable runtime_filter_mode=off
q1 5051 5096 5049 5049
q2 341 222 239 222
q3 4110 4000 3986 3986
q4 2815 2775 2767 2767
q5 9744 9695 9635 9635
q6 241 121 121 121
q7 3074 2521 2528 2521
q8 4915 4880 4821 4821
q9 13152 13042 13234 13042
q10 4267 4163 4215 4163
q11 774 644 692 644
q12 1016 807 819 807
q13 4277 3885 3874 3874
q14 362 362 358 358
q15 634 560 562 560
q16 769 745 717 717
q17 3907 3935 3939 3935
q18 9922 9408 9541 9408
q19 1796 1778 1773 1773
q20 2375 2065 2038 2038
q21 8932 9033 8673 8673
q22 965 863 849 849
Total cold run time: 83439 ms
Total hot run time: 79963 ms
```
</details>
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [WIP][Improve](Job)Refactor JOB [doris]
Posted by "doris-robot (via GitHub)" <gi...@apache.org>.
doris-robot commented on PR #26845:
URL: https://github.com/apache/doris/pull/26845#issuecomment-1808616857
(From new machine)TeamCity pipeline, clickbench performance test result:
the sum of best hot time: 46.97 seconds
stream load tsv: 553 seconds loaded 74807831229 Bytes, about 129 MB/s
stream load json: 20 seconds loaded 2358488459 Bytes, about 112 MB/s
stream load orc: 66 seconds loaded 1101869774 Bytes, about 15 MB/s
stream load parquet: 33 seconds loaded 861443392 Bytes, about 24 MB/s
insert into select: 28.9 seconds inserted 10000000 Rows, about 346K ops/s
storage size: 17162170002 Bytes
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [WIP][Improve](Job)Refactor JOB [doris]
Posted by "CalvinKirs (via GitHub)" <gi...@apache.org>.
CalvinKirs commented on PR #26845:
URL: https://github.com/apache/doris/pull/26845#issuecomment-1809389308
I commented out the metadata write so there will in memory for now, since the PR is large I recommend completing the optimization of the metadata in the next PR
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [WIP][Improve](Job)Refactor JOB [doris]
Posted by "CalvinKirs (via GitHub)" <gi...@apache.org>.
CalvinKirs commented on PR #26845:
URL: https://github.com/apache/doris/pull/26845#issuecomment-1808473585
run buildall
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [Improve](Job)Refactor JOB [doris]
Posted by "morningman (via GitHub)" <gi...@apache.org>.
morningman commented on code in PR #26845:
URL: https://github.com/apache/doris/pull/26845#discussion_r1391946138
##########
fe/fe-core/src/main/java/org/apache/doris/job/executor/DefaultTaskExecutorHandler.java:
##########
@@ -0,0 +1,75 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.job.executor;
+
+import org.apache.doris.job.disruptor.ExecuteTaskEvent;
+import org.apache.doris.job.manager.TaskTokenManager;
+import org.apache.doris.job.task.AbstractTask;
+
+import com.lmax.disruptor.WorkHandler;
+import lombok.extern.slf4j.Slf4j;
+
+import java.util.concurrent.Semaphore;
+
+/**
+ * DefaultTaskExecutor is an implementation of the TaskExecutor interface.
+ * if you need to implement your own TaskExecutor, you could refer to this class. and need to register
+ * it in the TaskExecutorFactory
+ * It executes a given AbstractTask by acquiring a semaphore token from the TaskTokenManager
+ * and releasing it after the task execution.
+ */
+@Slf4j
+public class DefaultTaskExecutorHandler<T extends AbstractTask> implements WorkHandler<ExecuteTaskEvent<T>> {
+
+
+ @Override
+ public void onEvent(ExecuteTaskEvent<T> executeTaskEvent) {
+ T task = executeTaskEvent.getTask();
+ if (null == task) {
+ log.warn("task is null, ignore,maybe task has been canceled");
+ return;
+ }
+ if (task.isCancelled()) {
+ log.info("task is canceled, ignore");
+ return;
+ }
+ if (null == executeTaskEvent.getJobConfig().getMaxConcurrentTaskNum()
+ || executeTaskEvent.getJobConfig().getMaxConcurrentTaskNum() <= 0) {
+ try {
+ task.runTask();
+ return;
+ } catch (Exception e) {
+ log.warn("execute task error, task id is {}", task.getTaskId(), e);
+ }
+ }
+ Semaphore semaphore = null;
Review Comment:
should be if..else here?
will it be blocked here?
##########
fe/fe-core/src/main/java/org/apache/doris/job/scheduler/JobScheduler.java:
##########
@@ -0,0 +1,173 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.job.scheduler;
+
+import org.apache.doris.common.CustomThreadFactory;
+import org.apache.doris.common.util.TimeUtils;
+import org.apache.doris.job.base.AbstractJob;
+import org.apache.doris.job.base.JobExecuteType;
+import org.apache.doris.job.common.JobStatus;
+import org.apache.doris.job.common.TaskType;
+import org.apache.doris.job.disruptor.TaskDisruptor;
+import org.apache.doris.job.executor.TimerJobSchedulerTask;
+import org.apache.doris.job.manager.TaskDisruptorGroupManager;
+import org.apache.doris.job.task.AbstractTask;
+
+import io.netty.util.HashedWheelTimer;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.collections.CollectionUtils;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+@Slf4j
+public class JobScheduler<T extends AbstractJob<?>> implements Closeable {
+
+ /**
+ * scheduler tasks, it's used to scheduler job
+ */
+ private HashedWheelTimer timerTaskScheduler;
+
+ private TaskDisruptor timerJobDisruptor;
+
+ private TaskDisruptorGroupManager taskDisruptorGroupManager;
+
+ private long latestBatchSchedulerTimerTaskTimeMs = 0L;
+
+ private static final long BATCH_SCHEDULER_INTERVAL_SECONDS = 60;
+
+ private static final int HASHED_WHEEL_TIMER_TICKS_PER_WHEEL = 660;
+
+ private final Map<Long, T> jobMap;
+
+ public JobScheduler(Map<Long, T> jobMap) {
+ this.jobMap = jobMap;
+ }
+
+ /**
+ * batch scheduler interval ms time
+ */
+ private static final long BATCH_SCHEDULER_INTERVAL_MILLI_SECONDS = BATCH_SCHEDULER_INTERVAL_SECONDS * 1000L;
+
+ public void start() {
+ timerTaskScheduler = new HashedWheelTimer(new CustomThreadFactory("timer-task-scheduler"), 1,
+ TimeUnit.SECONDS, HASHED_WHEEL_TIMER_TICKS_PER_WHEEL);
+ timerTaskScheduler.start();
+ taskDisruptorGroupManager = new TaskDisruptorGroupManager();
+ taskDisruptorGroupManager.init();
+ this.timerJobDisruptor = taskDisruptorGroupManager.getDispatchDisruptor();
+ latestBatchSchedulerTimerTaskTimeMs = System.currentTimeMillis();
+ batchSchedulerTimerJob();
+ cycleSystemSchedulerTasks();
+ }
+
+ /**
+ * We will cycle system scheduler tasks every 10 minutes.
+ * Jobs will be re-registered after the task is completed
+ */
+ private void cycleSystemSchedulerTasks() {
+ log.info("re-register system scheduler timer tasks" + TimeUtils.longToTimeString(System.currentTimeMillis()));
+ timerTaskScheduler.newTimeout(timeout -> {
+ batchSchedulerTimerJob();
+ cycleSystemSchedulerTasks();
+ }, BATCH_SCHEDULER_INTERVAL_SECONDS, TimeUnit.SECONDS);
+
+ }
+
+ private void batchSchedulerTimerJob() {
+ executeTimerJobIdsWithinLastTenMinutesWindow();
+ }
+
+ public void scheduleOneJob(T job) {
+ if (!job.getJobStatus().equals(JobStatus.RUNNING)) {
+ return;
+ }
+ if (!job.getJobConfig().checkIsTimerJob()) {
+ //manual job will not scheduler
+ if (JobExecuteType.MANUAL.equals(job.getJobConfig().getExecuteType())) {
+ return;
+ }
+ //todo skip streaming job,improve in the future
+ if (JobExecuteType.INSTANT.equals(job.getJobConfig().getExecuteType()) && job.isReadyForScheduling()) {
+ schedulerImmediateJob(job);
+ }
+ }
+ //if it's timer job and trigger last window already start, we will scheduler it immediately
+ cycleTimerJobScheduler(job);
+ }
+
+ @Override
+ public void close() throws IOException {
+
+ }
+
+
+ private void cycleTimerJobScheduler(T job) {
+ List<Long> delaySeconds = job.getJobConfig().getTriggerDelayTimes(System.currentTimeMillis(),
+ System.currentTimeMillis(), latestBatchSchedulerTimerTaskTimeMs);
+ if (CollectionUtils.isNotEmpty(delaySeconds)) {
+ delaySeconds.forEach(delaySecond -> {
+ TimerJobSchedulerTask<T> timerJobSchedulerTask = new TimerJobSchedulerTask<>(timerJobDisruptor, job);
+ timerTaskScheduler.newTimeout(timerJobSchedulerTask, delaySecond, TimeUnit.SECONDS);
+ });
+ }
+ }
+
+
+ private void schedulerImmediateJob(T job) {
+ List<? extends AbstractTask> tasks = job.createTasks(TaskType.MANUAL);
+ if (CollectionUtils.isEmpty(tasks)) {
+ return;
Review Comment:
Need to update the job's status?
Or the job will be in RUNNING state forever
##########
fe/fe-core/src/main/java/org/apache/doris/job/executor/TimerJobSchedulerTask.java:
##########
@@ -0,0 +1,48 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.job.executor;
+
+import org.apache.doris.job.base.AbstractJob;
+import org.apache.doris.job.disruptor.TaskDisruptor;
+
+import io.netty.util.Timeout;
+import io.netty.util.TimerTask;
+import jline.internal.Log;
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class TimerJobSchedulerTask<T extends AbstractJob<?>> implements TimerTask {
Review Comment:
What is the relation between TimerJobSchedulerTask and `Task`?
##########
fe/fe-core/src/main/java/org/apache/doris/analysis/CreateJobStmt.java:
##########
@@ -123,24 +112,43 @@ public void analyze(Analyzer analyzer) throws UserException {
labelName.analyze(analyzer);
String dbName = labelName.getDbName();
Env.getCurrentInternalCatalog().getDbOrAnalysisException(dbName);
- job.setDbName(labelName.getDbName());
- job.setJobName(labelName.getLabelName());
- if (StringUtils.isNotBlank(onceJobStartTimestamp)) {
- analyzerOnceTimeJob();
- } else {
- analyzerCycleJob();
+ analyzerSqlStmt();
+ // check its insert stmt,currently only support insert stmt
+ InsertJob job = new InsertJob();
+ JobExecutionConfiguration jobExecutionConfiguration = new JobExecutionConfiguration();
+ jobExecutionConfiguration.setExecuteType(executeType);
+ job.setCreateTimeMs(System.currentTimeMillis());
+ TimerDefinition timerDefinition = new TimerDefinition();
+
+ if (null != onceJobStartTimestamp) {
+ timerDefinition.setStartTimeMs(TimeUtils.timeStringToLong(onceJobStartTimestamp));
+ }
+ if (null != interval) {
+ timerDefinition.setInterval(interval);
+ }
+ if (null != intervalTimeUnit) {
+ timerDefinition.setIntervalUnit(IntervalUnit.valueOf(intervalTimeUnit.toUpperCase()));
}
- if (ConnectContext.get() != null) {
- timezone = ConnectContext.get().getSessionVariable().getTimeZone();
+ if (null != startsTimeStamp) {
+ timerDefinition.setStartTimeMs(TimeUtils.timeStringToLong(startsTimeStamp));
}
- timezone = TimeUtils.checkTimeZoneValidAndStandardize(timezone);
- job.setTimezone(timezone);
+ if (null != endsTimeStamp) {
+ timerDefinition.setEndTimeMs(TimeUtils.timeStringToLong(endsTimeStamp));
+ }
+ jobExecutionConfiguration.setTimerDefinition(timerDefinition);
+ job.setJobConfig(jobExecutionConfiguration);
+
job.setComment(comment);
- //todo support user define
- job.setUser(ConnectContext.get().getQualifiedUser());
+ job.setCurrentDbName(labelName.getDbName());
+ job.setJobName(labelName.getLabelName());
+ job.setComment(comment);
Review Comment:
duplicate
##########
fe/fe-core/src/main/java/org/apache/doris/job/executor/DefaultTaskExecutorHandler.java:
##########
@@ -0,0 +1,75 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.job.executor;
+
+import org.apache.doris.job.disruptor.ExecuteTaskEvent;
+import org.apache.doris.job.manager.TaskTokenManager;
+import org.apache.doris.job.task.AbstractTask;
+
+import com.lmax.disruptor.WorkHandler;
+import lombok.extern.slf4j.Slf4j;
+
+import java.util.concurrent.Semaphore;
+
+/**
+ * DefaultTaskExecutor is an implementation of the TaskExecutor interface.
+ * if you need to implement your own TaskExecutor, you could refer to this class. and need to register
+ * it in the TaskExecutorFactory
+ * It executes a given AbstractTask by acquiring a semaphore token from the TaskTokenManager
+ * and releasing it after the task execution.
+ */
+@Slf4j
+public class DefaultTaskExecutorHandler<T extends AbstractTask> implements WorkHandler<ExecuteTaskEvent<T>> {
+
+
+ @Override
+ public void onEvent(ExecuteTaskEvent<T> executeTaskEvent) {
+ T task = executeTaskEvent.getTask();
+ if (null == task) {
+ log.warn("task is null, ignore,maybe task has been canceled");
+ return;
+ }
+ if (task.isCancelled()) {
+ log.info("task is canceled, ignore");
+ return;
+ }
+ if (null == executeTaskEvent.getJobConfig().getMaxConcurrentTaskNum()
+ || executeTaskEvent.getJobConfig().getMaxConcurrentTaskNum() <= 0) {
+ try {
+ task.runTask();
+ return;
+ } catch (Exception e) {
+ log.warn("execute task error, task id is {}", task.getTaskId(), e);
Review Comment:
How to handle this exception?
I think the logic in `runTask()` can be moved to here?
##########
fe/fe-core/src/main/java/org/apache/doris/analysis/CreateJobStmt.java:
##########
@@ -123,24 +112,43 @@ public void analyze(Analyzer analyzer) throws UserException {
labelName.analyze(analyzer);
String dbName = labelName.getDbName();
Env.getCurrentInternalCatalog().getDbOrAnalysisException(dbName);
- job.setDbName(labelName.getDbName());
- job.setJobName(labelName.getLabelName());
- if (StringUtils.isNotBlank(onceJobStartTimestamp)) {
- analyzerOnceTimeJob();
- } else {
- analyzerCycleJob();
+ analyzerSqlStmt();
+ // check its insert stmt,currently only support insert stmt
+ InsertJob job = new InsertJob();
+ JobExecutionConfiguration jobExecutionConfiguration = new JobExecutionConfiguration();
+ jobExecutionConfiguration.setExecuteType(executeType);
+ job.setCreateTimeMs(System.currentTimeMillis());
+ TimerDefinition timerDefinition = new TimerDefinition();
+
+ if (null != onceJobStartTimestamp) {
+ timerDefinition.setStartTimeMs(TimeUtils.timeStringToLong(onceJobStartTimestamp));
+ }
+ if (null != interval) {
+ timerDefinition.setInterval(interval);
+ }
+ if (null != intervalTimeUnit) {
+ timerDefinition.setIntervalUnit(IntervalUnit.valueOf(intervalTimeUnit.toUpperCase()));
}
- if (ConnectContext.get() != null) {
- timezone = ConnectContext.get().getSessionVariable().getTimeZone();
+ if (null != startsTimeStamp) {
+ timerDefinition.setStartTimeMs(TimeUtils.timeStringToLong(startsTimeStamp));
}
- timezone = TimeUtils.checkTimeZoneValidAndStandardize(timezone);
- job.setTimezone(timezone);
+ if (null != endsTimeStamp) {
+ timerDefinition.setEndTimeMs(TimeUtils.timeStringToLong(endsTimeStamp));
+ }
+ jobExecutionConfiguration.setTimerDefinition(timerDefinition);
+ job.setJobConfig(jobExecutionConfiguration);
+
job.setComment(comment);
- //todo support user define
- job.setUser(ConnectContext.get().getQualifiedUser());
+ job.setCurrentDbName(labelName.getDbName());
+ job.setJobName(labelName.getLabelName());
+ job.setComment(comment);
+ job.setCreateUser(ConnectContext.get().getQualifiedUser());
Review Comment:
```suggestion
job.setCreateUser(ConnectContext.get().getCurrentUserIdentity());
```
##########
fe/fe-core/src/main/java/org/apache/doris/job/scheduler/JobScheduler.java:
##########
@@ -0,0 +1,171 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.job.scheduler;
+
+import org.apache.doris.common.CustomThreadFactory;
+import org.apache.doris.common.util.TimeUtils;
+import org.apache.doris.job.base.AbstractJob;
+import org.apache.doris.job.base.JobExecuteType;
+import org.apache.doris.job.common.JobStatus;
+import org.apache.doris.job.common.TaskType;
+import org.apache.doris.job.disruptor.TaskDisruptor;
+import org.apache.doris.job.executor.TimerJobSchedulerTask;
+import org.apache.doris.job.manager.TaskDisruptorGroupManager;
+import org.apache.doris.job.task.AbstractTask;
+
+import io.netty.util.HashedWheelTimer;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.collections.CollectionUtils;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+@Slf4j
+public class JobScheduler<T extends AbstractJob<?>> implements Closeable {
Review Comment:
So you need a AbstractJobScheduler
##########
fe/fe-core/src/main/java/org/apache/doris/analysis/CreateJobStmt.java:
##########
@@ -81,16 +81,17 @@ public class CreateJobStmt extends DdlStmt {
private final String endsTimeStamp;
private final String comment;
+ private JobExecuteType executeType;
private String timezone = TimeUtils.DEFAULT_TIME_ZONE;
private static final ImmutableSet<Class<? extends DdlStmt>> supportStmtSuperClass
= new ImmutableSet.Builder<Class<? extends DdlStmt>>().add(InsertStmt.class)
Review Comment:
Use InsertIntoCommand
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [WIP][Improve](Job)Refactor JOB [doris]
Posted by "CalvinKirs (via GitHub)" <gi...@apache.org>.
CalvinKirs commented on code in PR #26845:
URL: https://github.com/apache/doris/pull/26845#discussion_r1390930719
##########
fe/fe-core/src/main/java/org/apache/doris/job/manager/JobManager.java:
##########
@@ -0,0 +1,213 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.job.manager;
+
+import org.apache.doris.catalog.Env;
+import org.apache.doris.common.io.Writable;
+import org.apache.doris.common.util.LogBuilder;
+import org.apache.doris.common.util.LogKey;
+import org.apache.doris.job.base.AbstractJob;
+import org.apache.doris.job.common.JobStatus;
+import org.apache.doris.job.common.JobType;
+import org.apache.doris.job.exception.JobException;
+import org.apache.doris.job.scheduler.JobScheduler;
+import org.apache.doris.job.task.AbstractTask;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+import java.util.List;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.stream.Collectors;
+
+@Slf4j
+public class JobManager<T extends AbstractJob<?>> implements Writable {
Review Comment:
Manager exposes too many of its own dedicated interfaces, such as reading metadata, etc. Maybe it should extract a separate registration interface
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [Improve](Job)Refactor JOB [doris]
Posted by "CalvinKirs (via GitHub)" <gi...@apache.org>.
CalvinKirs commented on PR #26845:
URL: https://github.com/apache/doris/pull/26845#issuecomment-1809634324
run buildall
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [WIP][Improve](Job)Refactor JOB [doris]
Posted by "CalvinKirs (via GitHub)" <gi...@apache.org>.
CalvinKirs commented on PR #26845:
URL: https://github.com/apache/doris/pull/26845#issuecomment-1807987464
run FE UT (Doris FE UT)
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [Improve](Job)Refactor JOB [doris]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #26845:
URL: https://github.com/apache/doris/pull/26845#issuecomment-1809827533
PR approved by at least one committer and no changes requested.
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [WIP][Improve](Job)Refactor JOB [doris]
Posted by "CalvinKirs (via GitHub)" <gi...@apache.org>.
CalvinKirs commented on PR #26845:
URL: https://github.com/apache/doris/pull/26845#issuecomment-1807910609
run buildall
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [WIP][Improve](Job)Refactor JOB [doris]
Posted by "CalvinKirs (via GitHub)" <gi...@apache.org>.
CalvinKirs commented on PR #26845:
URL: https://github.com/apache/doris/pull/26845#issuecomment-1807935040
run buildall
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [WIP][Improve](Job)Refactor JOB [doris]
Posted by "morningman (via GitHub)" <gi...@apache.org>.
morningman commented on code in PR #26845:
URL: https://github.com/apache/doris/pull/26845#discussion_r1391106633
##########
fe/fe-core/src/main/java/org/apache/doris/job/base/Job.java:
##########
@@ -0,0 +1,60 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.job.base;
+
+import org.apache.doris.job.common.JobType;
+import org.apache.doris.job.common.TaskType;
+import org.apache.doris.job.exception.JobException;
+import org.apache.doris.job.task.AbstractTask;
+import org.apache.doris.qe.ShowResultSetMetaData;
+
+import java.util.List;
+
+/**
+ * Job is the core of the scheduler module, which is used to store the Job information of the job module.
+ * We can use the job to uniquely identify a Job.
+ * The jobName is used to identify the job, which is not unique.
+ * The jobStatus is used to identify the status of the Job, which is used to control the execution of the
+ * job.
+ */
+public interface Job<T extends AbstractTask> {
+
+ List<T> createTasks(TaskType taskType);
Review Comment:
Add comments for each interface' method
##########
fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/InsertTask.java:
##########
@@ -0,0 +1,81 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.job.extensions.insert;
+
+import org.apache.doris.common.util.TimeUtils;
+import org.apache.doris.job.task.AbstractTask;
+import org.apache.doris.load.FailMsg;
+import org.apache.doris.load.loadv2.LoadJob;
+import org.apache.doris.nereids.trees.plans.commands.InsertIntoTableCommand;
+import org.apache.doris.qe.ShowResultSetMetaData;
+
+import lombok.extern.slf4j.Slf4j;
+
+
+@Slf4j
+public class InsertTask extends AbstractTask {
+
+ private String labelName;
+
+ private InsertIntoTableCommand command;
+ private LoadJob.LoadStatistic statistic;
+ private FailMsg failMsg;
+
+ private InsertIntoState insertIntoState;
+
+ @Override
+ public void before() {
+ super.before();
+ }
+
+ public InsertTask(String labelName, InsertIntoTableCommand command, LoadJob.LoadStatistic statistic,
+ FailMsg failMsg, InsertIntoState insertIntoState) {
+ this.labelName = labelName;
+ this.command = command;
+ this.statistic = statistic;
+ this.failMsg = failMsg;
+ this.insertIntoState = insertIntoState;
+ }
+
+ @Override
+ public void run() {
+ //insertIntoState = command.
+ System.out.println(getJobId() + "InsertTask run" + TimeUtils.longToTimeString(System.currentTimeMillis()));
Review Comment:
`System.out.println`?
##########
fe/fe-core/src/main/java/org/apache/doris/job/manager/TaskTokenManager.java:
##########
@@ -0,0 +1,57 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.job.manager;
+
+import jline.internal.Log;
Review Comment:
What is this?
##########
fe/fe-core/src/main/java/org/apache/doris/job/manager/JobManager.java:
##########
@@ -0,0 +1,213 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.job.manager;
+
+import org.apache.doris.catalog.Env;
+import org.apache.doris.common.io.Writable;
+import org.apache.doris.common.util.LogBuilder;
+import org.apache.doris.common.util.LogKey;
+import org.apache.doris.job.base.AbstractJob;
+import org.apache.doris.job.common.JobStatus;
+import org.apache.doris.job.common.JobType;
+import org.apache.doris.job.exception.JobException;
+import org.apache.doris.job.scheduler.JobScheduler;
+import org.apache.doris.job.task.AbstractTask;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+import java.util.List;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.stream.Collectors;
+
+@Slf4j
+public class JobManager<T extends AbstractJob<?>> implements Writable {
Review Comment:
Do we need to use template for JobManager？
##########
fe/fe-core/src/main/java/org/apache/doris/job/base/AbstractJob.java:
##########
@@ -0,0 +1,185 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.job.base;
+
+import org.apache.doris.catalog.Env;
+import org.apache.doris.common.io.Text;
+import org.apache.doris.common.io.Writable;
+import org.apache.doris.common.util.TimeUtils;
+import org.apache.doris.job.common.JobStatus;
+import org.apache.doris.job.common.JobType;
+import org.apache.doris.job.common.TaskStatus;
+import org.apache.doris.job.exception.JobException;
+import org.apache.doris.job.extensions.insert.InsertJob;
+import org.apache.doris.job.task.AbstractTask;
+import org.apache.doris.job.task.Task;
+
+import com.google.gson.annotations.SerializedName;
+import lombok.Data;
+import org.apache.commons.collections.CollectionUtils;
+
+import java.io.DataInput;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
+@Data
+public abstract class AbstractJob<T extends AbstractTask> implements Job<T>, Writable {
+
+ @SerializedName(value = "jobId")
+ private Long jobId;
+
+ @SerializedName(value = "jobName")
+ private String jobName;
+
+ @SerializedName(value = "jobStatus")
+ private JobStatus jobStatus;
+
+ @SerializedName(value = "currentDbName")
+ private String currentDbName;
+
+ @SerializedName(value = "comment")
+ private String comment;
+
+ @SerializedName(value = "jobType")
+ private String createUser;
+
+ @SerializedName(value = "jobConfig")
+ private JobExecutionConfiguration jobConfig;
+
+ @SerializedName(value = "createTimeMs")
+ private Long createTimeMs;
+
+ @SerializedName(value = "executeSql")
+ String executeSql;
+
+ private List<T> runningTasks = new ArrayList<>();
+
+ @Override
+ public void cancel() throws JobException {
+ if (CollectionUtils.isEmpty(runningTasks)) {
+ return;
+ }
+ runningTasks.forEach(Task::cancel);
+
+ }
+
+ public void initTasks(List<T> tasks) {
+ tasks.forEach(task -> {
+ task.setJobId(jobId);
+ task.setTaskId(Env.getCurrentEnv().getNextId());
Review Comment:
`Env.getCurrentEnv().getNextId()` may write edit log,
So it need to be very carefully to call `initTasks()`.
I think we should find a better to way to init tasks, eg, without calling `Env.getCurrentEnv().getNextId()` inside
##########
fe/fe-core/src/main/java/org/apache/doris/job/base/AbstractJob.java:
##########
@@ -0,0 +1,185 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.job.base;
+
+import org.apache.doris.catalog.Env;
+import org.apache.doris.common.io.Text;
+import org.apache.doris.common.io.Writable;
+import org.apache.doris.common.util.TimeUtils;
+import org.apache.doris.job.common.JobStatus;
+import org.apache.doris.job.common.JobType;
+import org.apache.doris.job.common.TaskStatus;
+import org.apache.doris.job.exception.JobException;
+import org.apache.doris.job.extensions.insert.InsertJob;
+import org.apache.doris.job.task.AbstractTask;
+import org.apache.doris.job.task.Task;
+
+import com.google.gson.annotations.SerializedName;
+import lombok.Data;
+import org.apache.commons.collections.CollectionUtils;
+
+import java.io.DataInput;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
+@Data
+public abstract class AbstractJob<T extends AbstractTask> implements Job<T>, Writable {
+
+ @SerializedName(value = "jobId")
+ private Long jobId;
+
+ @SerializedName(value = "jobName")
+ private String jobName;
+
+ @SerializedName(value = "jobStatus")
+ private JobStatus jobStatus;
+
+ @SerializedName(value = "currentDbName")
+ private String currentDbName;
+
+ @SerializedName(value = "comment")
+ private String comment;
+
+ @SerializedName(value = "jobType")
+ private String createUser;
+
+ @SerializedName(value = "jobConfig")
+ private JobExecutionConfiguration jobConfig;
+
+ @SerializedName(value = "createTimeMs")
+ private Long createTimeMs;
+
+ @SerializedName(value = "executeSql")
+ String executeSql;
+
+ private List<T> runningTasks = new ArrayList<>();
+
+ @Override
+ public void cancel() throws JobException {
+ if (CollectionUtils.isEmpty(runningTasks)) {
+ return;
+ }
+ runningTasks.forEach(Task::cancel);
+
+ }
+
+ public void initTasks(List<T> tasks) {
+ tasks.forEach(task -> {
+ task.setJobId(jobId);
+ task.setTaskId(Env.getCurrentEnv().getNextId());
+ task.setCreateTimeMs(System.currentTimeMillis());
+ task.setStatus(TaskStatus.PENDING);
+ });
+ }
+
+ public void checkJobParams() {
+ if (null == jobConfig) {
+ throw new IllegalArgumentException("jobConfig cannot be null");
+ }
+ jobConfig.checkParams(createTimeMs);
+ checkJobParamsInternal();
+ }
+
+ public void updateJobStatus(JobStatus newJobStatus) {
+ if (null == newJobStatus) {
+ throw new IllegalArgumentException("jobStatus cannot be null");
+ }
+ if (jobStatus == newJobStatus) {
+ throw new IllegalArgumentException(String.format("Can't update job %s status to the %s status",
+ jobStatus.name(), this.jobStatus.name()));
+ }
+ if (newJobStatus.equals(JobStatus.RUNNING) && !jobStatus.equals(JobStatus.PAUSED)) {
+ throw new IllegalArgumentException(String.format("Can't update job %s status to the %s status",
+ jobStatus.name(), this.jobStatus.name()));
+ }
+ if (newJobStatus.equals(JobStatus.STOPPED) && !jobStatus.equals(JobStatus.RUNNING)) {
+ throw new IllegalArgumentException(String.format("Can't update job %s status to the %s status",
+ jobStatus.name(), this.jobStatus.name()));
+ }
+ jobStatus = newJobStatus;
+ }
+
+
+ protected abstract void checkJobParamsInternal();
+
+ public static AbstractJob readFields(DataInput in) throws IOException {
+ JobType jobType = JobType.valueOf(Text.readString(in));
Review Comment:
Better use `RuntimeTypeAdapterFactory` of Gson to do the serde
##########
fe/fe-core/src/main/java/org/apache/doris/job/base/Job.java:
##########
@@ -0,0 +1,60 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.job.base;
+
+import org.apache.doris.job.common.JobType;
+import org.apache.doris.job.common.TaskType;
+import org.apache.doris.job.exception.JobException;
+import org.apache.doris.job.task.AbstractTask;
+import org.apache.doris.qe.ShowResultSetMetaData;
+
+import java.util.List;
+
+/**
+ * Job is the core of the scheduler module, which is used to store the Job information of the job module.
+ * We can use the job to uniquely identify a Job.
+ * The jobName is used to identify the job, which is not unique.
+ * The jobStatus is used to identify the status of the Job, which is used to control the execution of the
+ * job.
+ */
+public interface Job<T extends AbstractTask> {
+
+ List<T> createTasks(TaskType taskType);
+
+ void cancel(T task) throws JobException;
+
+ boolean isReadyForScheduling();
+
+
+ ShowResultSetMetaData getJobMetaData();
+
+ ShowResultSetMetaData getTaskMetaData();
+
+ JobType getJobType();
+
+ List<T> queryTasks();
+
+ void cancel() throws JobException;
+
+ void onTaskFail(T task);
Review Comment:
Should it be the `Task`'s methods?
##########
fe/fe-core/src/main/java/org/apache/doris/job/base/AbstractJob.java:
##########
@@ -0,0 +1,185 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.job.base;
+
+import org.apache.doris.catalog.Env;
+import org.apache.doris.common.io.Text;
+import org.apache.doris.common.io.Writable;
+import org.apache.doris.common.util.TimeUtils;
+import org.apache.doris.job.common.JobStatus;
+import org.apache.doris.job.common.JobType;
+import org.apache.doris.job.common.TaskStatus;
+import org.apache.doris.job.exception.JobException;
+import org.apache.doris.job.extensions.insert.InsertJob;
+import org.apache.doris.job.task.AbstractTask;
+import org.apache.doris.job.task.Task;
+
+import com.google.gson.annotations.SerializedName;
+import lombok.Data;
+import org.apache.commons.collections.CollectionUtils;
+
+import java.io.DataInput;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
+@Data
+public abstract class AbstractJob<T extends AbstractTask> implements Job<T>, Writable {
+
+ @SerializedName(value = "jobId")
Review Comment:
Use abbr for the `SerializedName`, such as `jid` for `jobId`
##########
fe/fe-core/src/main/java/org/apache/doris/job/base/Job.java:
##########
@@ -0,0 +1,60 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.job.base;
+
+import org.apache.doris.job.common.JobType;
+import org.apache.doris.job.common.TaskType;
+import org.apache.doris.job.exception.JobException;
+import org.apache.doris.job.task.AbstractTask;
+import org.apache.doris.qe.ShowResultSetMetaData;
+
+import java.util.List;
+
+/**
+ * Job is the core of the scheduler module, which is used to store the Job information of the job module.
+ * We can use the job to uniquely identify a Job.
+ * The jobName is used to identify the job, which is not unique.
+ * The jobStatus is used to identify the status of the Job, which is used to control the execution of the
+ * job.
+ */
+public interface Job<T extends AbstractTask> {
+
+ List<T> createTasks(TaskType taskType);
+
+ void cancel(T task) throws JobException;
Review Comment:
better pass task id?
##########
fe/fe-core/src/main/java/org/apache/doris/job/manager/JobManager.java:
##########
@@ -0,0 +1,213 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.job.manager;
+
+import org.apache.doris.catalog.Env;
+import org.apache.doris.common.io.Writable;
+import org.apache.doris.common.util.LogBuilder;
+import org.apache.doris.common.util.LogKey;
+import org.apache.doris.job.base.AbstractJob;
+import org.apache.doris.job.common.JobStatus;
+import org.apache.doris.job.common.JobType;
+import org.apache.doris.job.exception.JobException;
+import org.apache.doris.job.scheduler.JobScheduler;
+import org.apache.doris.job.task.AbstractTask;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+import java.util.List;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.stream.Collectors;
+
+@Slf4j
+public class JobManager<T extends AbstractJob<?>> implements Writable {
+
+
+ private final ConcurrentHashMap<Long, T> jobMap = new ConcurrentHashMap<>(32);
+
+ private JobScheduler jobScheduler;
+
+ public void start() {
+ jobScheduler = new JobScheduler(jobMap);
+ jobScheduler.start();
+ }
+
+ public Long registerJob(T job) throws JobException {
+ //job.checkJobParams();
+ checkJobNameExist(job.getJobName(), job.getJobType(), job.getCurrentDbName());
+ long id = Env.getCurrentEnv().getNextId();
+ job.setJobId(id);
+ Env.getCurrentEnv().getEditLog().logCreateJob(job);
+ //check name exist
+ jobMap.put(id, job);
+ //check its need to scheduler
+ jobScheduler.scheduleOneJob(job);
+ return id;
+ }
+
+
+ private void checkJobNameExist(String jobName, JobType type, String currentDbName) throws JobException {
+ if (jobMap.values().stream().anyMatch(a -> a.getJobName().equals(jobName) && a.getJobType().equals(type)
+ && (null == a.getCurrentDbName() || a.getCurrentDbName().equals(currentDbName)))) {
+ throw new JobException("job name exist,jobName:" + jobName);
+ }
+ }
+
+ public void unregisterJob(Long jobId) throws JobException {
+ checkJobExist(jobId);
+ jobMap.get(jobId).setJobStatus(JobStatus.STOPPED);
+ jobMap.get(jobId).cancel();
+ Env.getCurrentEnv().getEditLog().logDeleteJob(jobMap.get(jobId));
+ jobMap.remove(jobId);
+ }
+
+ public void unregisterJob(String currentDbName, String jobName) throws JobException {
+ for (T a : jobMap.values()) {
+ if (a.getJobName().equals(jobName) && (null != a.getCurrentDbName()
+ && a.getCurrentDbName().equals(currentDbName)) && a.getJobType().equals(JobType.INSERT)) {
+ try {
+ unregisterJob(a.getJobId());
+ } catch (JobException e) {
+ throw new JobException("unregister job error,jobName:" + jobName);
+ }
+ }
+ }
+
+ }
+
+ public void alterJobStatus(Long jobId, JobStatus status) throws JobException {
+ checkJobExist(jobId);
+ jobMap.get(jobId).updateJobStatus(status);
+ Env.getCurrentEnv().getEditLog().logUpdateJob(jobMap.get(jobId));
+ }
+
+ public void alterJobStatus(String currentDbName, String jobName, JobStatus jobStatus) throws JobException {
+ for (T a : jobMap.values()) {
+ if (a.getJobName().equals(jobName) && (null != a.getCurrentDbName()
+ && a.getCurrentDbName().equals(currentDbName)) && JobType.INSERT.equals(a.getJobType())) {
+ try {
+ alterJobStatus(a.getJobId(), jobStatus);
+ } catch (JobException e) {
+ throw new JobException("unregister job error,jobName:" + jobName);
+ }
+ }
+ }
+ }
+
+ private void checkJobExist(Long jobId) throws JobException {
+ if (null == jobMap.get(jobId)) {
+ throw new JobException("job not exist,jobId:" + jobId);
+ }
+ }
+
+ public List<T> queryJobs(JobType type) {
+ return jobMap.values().stream().filter(a -> a.getJobType().equals(type))
+ .collect(java.util.stream.Collectors.toList());
+ }
+
+ public List<T> queryJobs(String currentDb, String jobName) {
+ //only query insert job,we just provide insert job
+ return jobMap.values().stream().filter(a -> checkItsMatch(currentDb, jobName, a))
+ .collect(Collectors.toList());
+ }
+
+ private boolean checkItsMatch(String currentDb, String jobName, T job) {
+ if (StringUtils.isBlank(jobName)) {
+ return job.getJobType().equals(JobType.INSERT) && (null != job.getCurrentDbName()
+ && job.getCurrentDbName().equals(currentDb));
+ }
+ return job.getJobType().equals(JobType.INSERT) && (null != job.getCurrentDbName()
+ && job.getCurrentDbName().equals(currentDb)) && job.getJobName().equals(jobName);
+ }
+
+ public List<? extends AbstractTask> queryTasks(Long jobId) throws JobException {
+ checkJobExist(jobId);
+ return jobMap.get(jobId).queryTasks();
+ }
+
+ public void replayCreateJob(T job) {
+ if (jobMap.containsKey(job.getJobId())) {
+ return;
+ }
+ jobMap.putIfAbsent(job.getJobId(), job);
+ log.info(new LogBuilder(LogKey.SCHEDULER_JOB, job.getJobId())
+ .add("msg", "replay create scheduler job").build());
+ }
+
+ /**
+ * Replay update load job.
+ **/
+ public void replayUpdateJob(T job) {
+ jobMap.put(job.getJobId(), job);
+ log.info(new LogBuilder(LogKey.SCHEDULER_JOB, job.getJobId())
+ .add("msg", "replay update scheduler job").build());
+ }
+
+ public void replayDeleteJob(T job) {
+ if (null == jobMap.get(job.getJobId())) {
+ return;
+ }
+ jobMap.remove(job.getJobId());
+ log.info(new LogBuilder(LogKey.SCHEDULER_JOB, job.getJobId())
+ .add("msg", "replay delete scheduler job").build());
+ }
+
+ void cancelTask(Long jobId, Long taskId) throws JobException {
+ checkJobExist(jobId);
+ if (null == jobMap.get(jobId).getRunningTasks()) {
+ throw new JobException("task not exist,taskId:" + taskId);
+ }
+ }
+
+ @Override
+ public void write(DataOutput out) throws IOException {
+ out.writeInt(jobMap.size());
Review Comment:
Use Gson
##########
fe/fe-core/src/main/java/org/apache/doris/job/manager/JobManager.java:
##########
@@ -0,0 +1,213 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.job.manager;
+
+import org.apache.doris.catalog.Env;
+import org.apache.doris.common.io.Writable;
+import org.apache.doris.common.util.LogBuilder;
+import org.apache.doris.common.util.LogKey;
+import org.apache.doris.job.base.AbstractJob;
+import org.apache.doris.job.common.JobStatus;
+import org.apache.doris.job.common.JobType;
+import org.apache.doris.job.exception.JobException;
+import org.apache.doris.job.scheduler.JobScheduler;
+import org.apache.doris.job.task.AbstractTask;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+import java.util.List;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.stream.Collectors;
+
+@Slf4j
+public class JobManager<T extends AbstractJob<?>> implements Writable {
+
+
+ private final ConcurrentHashMap<Long, T> jobMap = new ConcurrentHashMap<>(32);
+
+ private JobScheduler jobScheduler;
+
+ public void start() {
+ jobScheduler = new JobScheduler(jobMap);
+ jobScheduler.start();
+ }
+
+ public Long registerJob(T job) throws JobException {
+ //job.checkJobParams();
+ checkJobNameExist(job.getJobName(), job.getJobType(), job.getCurrentDbName());
+ long id = Env.getCurrentEnv().getNextId();
Review Comment:
I suggest that the job id should be assigned when job is created.
Not in `registerJob()` method, so it will easier to write unit test
##########
fe/fe-core/src/main/java/org/apache/doris/job/manager/TaskDisruptorGroupManager.java:
##########
@@ -0,0 +1,96 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.job.manager;
+
+import org.apache.doris.common.CustomThreadFactory;
+import org.apache.doris.job.base.AbstractJob;
+import org.apache.doris.job.base.JobExecutionConfiguration;
+import org.apache.doris.job.common.JobType;
+import org.apache.doris.job.disruptor.ExecuteTaskEvent;
+import org.apache.doris.job.disruptor.TaskDisruptor;
+import org.apache.doris.job.disruptor.TimerJobEvent;
+import org.apache.doris.job.executor.DefaultTaskExecutorHandler;
+import org.apache.doris.job.executor.DispatchTaskHandler;
+import org.apache.doris.job.extensions.insert.InsertTask;
+import org.apache.doris.job.task.AbstractTask;
+
+import com.lmax.disruptor.BlockingWaitStrategy;
+import com.lmax.disruptor.EventFactory;
+import com.lmax.disruptor.EventTranslatorVararg;
+import com.lmax.disruptor.WorkHandler;
+import lombok.Getter;
+
+import java.util.EnumMap;
+import java.util.Map;
+import java.util.concurrent.ThreadFactory;
+
+public class TaskDisruptorGroupManager<T extends AbstractTask> {
+
+ private final Map<JobType, TaskDisruptor<T>> disruptorMap = new EnumMap<>(JobType.class);
+
+ @Getter
+ private TaskDisruptor<TimerJobEvent<AbstractJob<?>>> dispatchDisruptor;
+
+
+ public void init() {
+ registerInsertDisruptor();
+ //when all task queue is ready, dispatch task to registered task executor
+ registerDispatchDisruptor();
+ }
+
+ private void registerDispatchDisruptor() {
+ EventFactory<TimerJobEvent<AbstractJob<T>>> dispatchEventFactory = TimerJobEvent.factory();
+ ThreadFactory dispatchThreadFactory = new CustomThreadFactory("dispatch-task");
+ WorkHandler[] dispatchTaskExecutorHandlers = new WorkHandler[5];
+ for (int i = 0; i < 5; i++) {
+ dispatchTaskExecutorHandlers[i] = new DispatchTaskHandler(this.disruptorMap);
+ }
+ EventTranslatorVararg<TimerJobEvent<AbstractJob<T>>> eventTranslator =
+ (event, sequence, args) -> event.setJob((AbstractJob<T>) args[0]);
+ this.dispatchDisruptor = new TaskDisruptor<>(dispatchEventFactory, 1024, dispatchThreadFactory,
+ new BlockingWaitStrategy(), dispatchTaskExecutorHandlers, eventTranslator);
+ }
+
+ private void registerInsertDisruptor() {
+ EventFactory<ExecuteTaskEvent<InsertTask>> insertEventFactory = ExecuteTaskEvent.factory();
+ ThreadFactory insertTaskThreadFactory = new CustomThreadFactory("insert-task-execute");
+ WorkHandler[] insertTaskExecutorHandlers = new WorkHandler[5];
+ for (int i = 0; i < 5; i++) {
+ insertTaskExecutorHandlers[i] = new DefaultTaskExecutorHandler<InsertTask>();
+ }
+ EventTranslatorVararg<ExecuteTaskEvent<InsertTask>> eventTranslator =
+ (event, sequence, args) -> {
+ event.setTask((InsertTask) args[0]);
+ event.setJobConfig((JobExecutionConfiguration) args[1]);
+ };
+ TaskDisruptor insertDisruptor = new TaskDisruptor<>(insertEventFactory, 1024,
Review Comment:
1024 is a magic number
##########
fe/fe-core/src/main/java/org/apache/doris/job/scheduler/JobScheduler.java:
##########
@@ -0,0 +1,171 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.job.scheduler;
+
+import org.apache.doris.common.CustomThreadFactory;
+import org.apache.doris.common.util.TimeUtils;
+import org.apache.doris.job.base.AbstractJob;
+import org.apache.doris.job.base.JobExecuteType;
+import org.apache.doris.job.common.JobStatus;
+import org.apache.doris.job.common.TaskType;
+import org.apache.doris.job.disruptor.TaskDisruptor;
+import org.apache.doris.job.executor.TimerJobSchedulerTask;
+import org.apache.doris.job.manager.TaskDisruptorGroupManager;
+import org.apache.doris.job.task.AbstractTask;
+
+import io.netty.util.HashedWheelTimer;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.collections.CollectionUtils;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+@Slf4j
+public class JobScheduler<T extends AbstractJob<?>> implements Closeable {
Review Comment:
Why using template?
##########
fe/fe-core/src/main/java/org/apache/doris/job/scheduler/JobScheduler.java:
##########
@@ -0,0 +1,171 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.job.scheduler;
+
+import org.apache.doris.common.CustomThreadFactory;
+import org.apache.doris.common.util.TimeUtils;
+import org.apache.doris.job.base.AbstractJob;
+import org.apache.doris.job.base.JobExecuteType;
+import org.apache.doris.job.common.JobStatus;
+import org.apache.doris.job.common.TaskType;
+import org.apache.doris.job.disruptor.TaskDisruptor;
+import org.apache.doris.job.executor.TimerJobSchedulerTask;
+import org.apache.doris.job.manager.TaskDisruptorGroupManager;
+import org.apache.doris.job.task.AbstractTask;
+
+import io.netty.util.HashedWheelTimer;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.collections.CollectionUtils;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+@Slf4j
+public class JobScheduler<T extends AbstractJob<?>> implements Closeable {
+
+ /**
+ * scheduler tasks, it's used to scheduler job
+ */
+ private HashedWheelTimer timerTaskScheduler;
+
+ private TaskDisruptor timerJobDisruptor;
+
+ private TaskDisruptorGroupManager taskDisruptorGroupManager;
+
+ private long latestBatchSchedulerTimerTaskTimeMs = 0L;
+
+ private static final long BATCH_SCHEDULER_INTERVAL_SECONDS = 60;
+
+ private final Map<Long, T> jobMap;
+
+ public JobScheduler(Map<Long, T> jobMap) {
+ this.jobMap = jobMap;
+ }
+
+ /**
+ * batch scheduler interval ms time
+ */
+ private static final long BATCH_SCHEDULER_INTERVAL_MILLI_SECONDS = BATCH_SCHEDULER_INTERVAL_SECONDS * 1000L;
+
+ public void start() {
+ timerTaskScheduler = new HashedWheelTimer(new CustomThreadFactory("timer-task-scheduler"), 1,
+ TimeUnit.SECONDS, 660);
Review Comment:
660 is a magic number
##########
fe/fe-core/src/main/java/org/apache/doris/job/manager/TaskDisruptorGroupManager.java:
##########
@@ -0,0 +1,96 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.job.manager;
+
+import org.apache.doris.common.CustomThreadFactory;
+import org.apache.doris.job.base.AbstractJob;
+import org.apache.doris.job.base.JobExecutionConfiguration;
+import org.apache.doris.job.common.JobType;
+import org.apache.doris.job.disruptor.ExecuteTaskEvent;
+import org.apache.doris.job.disruptor.TaskDisruptor;
+import org.apache.doris.job.disruptor.TimerJobEvent;
+import org.apache.doris.job.executor.DefaultTaskExecutorHandler;
+import org.apache.doris.job.executor.DispatchTaskHandler;
+import org.apache.doris.job.extensions.insert.InsertTask;
+import org.apache.doris.job.task.AbstractTask;
+
+import com.lmax.disruptor.BlockingWaitStrategy;
+import com.lmax.disruptor.EventFactory;
+import com.lmax.disruptor.EventTranslatorVararg;
+import com.lmax.disruptor.WorkHandler;
+import lombok.Getter;
+
+import java.util.EnumMap;
+import java.util.Map;
+import java.util.concurrent.ThreadFactory;
+
+public class TaskDisruptorGroupManager<T extends AbstractTask> {
+
+ private final Map<JobType, TaskDisruptor<T>> disruptorMap = new EnumMap<>(JobType.class);
+
+ @Getter
+ private TaskDisruptor<TimerJobEvent<AbstractJob<?>>> dispatchDisruptor;
+
+
+ public void init() {
+ registerInsertDisruptor();
+ //when all task queue is ready, dispatch task to registered task executor
+ registerDispatchDisruptor();
+ }
+
+ private void registerDispatchDisruptor() {
+ EventFactory<TimerJobEvent<AbstractJob<T>>> dispatchEventFactory = TimerJobEvent.factory();
+ ThreadFactory dispatchThreadFactory = new CustomThreadFactory("dispatch-task");
+ WorkHandler[] dispatchTaskExecutorHandlers = new WorkHandler[5];
+ for (int i = 0; i < 5; i++) {
+ dispatchTaskExecutorHandlers[i] = new DispatchTaskHandler(this.disruptorMap);
+ }
+ EventTranslatorVararg<TimerJobEvent<AbstractJob<T>>> eventTranslator =
+ (event, sequence, args) -> event.setJob((AbstractJob<T>) args[0]);
+ this.dispatchDisruptor = new TaskDisruptor<>(dispatchEventFactory, 1024, dispatchThreadFactory,
+ new BlockingWaitStrategy(), dispatchTaskExecutorHandlers, eventTranslator);
+ }
+
+ private void registerInsertDisruptor() {
+ EventFactory<ExecuteTaskEvent<InsertTask>> insertEventFactory = ExecuteTaskEvent.factory();
+ ThreadFactory insertTaskThreadFactory = new CustomThreadFactory("insert-task-execute");
+ WorkHandler[] insertTaskExecutorHandlers = new WorkHandler[5];
+ for (int i = 0; i < 5; i++) {
Review Comment:
5 is a magic number
##########
fe/fe-core/src/main/java/org/apache/doris/job/manager/JobManager.java:
##########
@@ -0,0 +1,213 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.job.manager;
+
+import org.apache.doris.catalog.Env;
+import org.apache.doris.common.io.Writable;
+import org.apache.doris.common.util.LogBuilder;
+import org.apache.doris.common.util.LogKey;
+import org.apache.doris.job.base.AbstractJob;
+import org.apache.doris.job.common.JobStatus;
+import org.apache.doris.job.common.JobType;
+import org.apache.doris.job.exception.JobException;
+import org.apache.doris.job.scheduler.JobScheduler;
+import org.apache.doris.job.task.AbstractTask;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+import java.util.List;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.stream.Collectors;
+
+@Slf4j
+public class JobManager<T extends AbstractJob<?>> implements Writable {
Review Comment:
Do we need to use template for JobManager？
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [Improve](Job)Refactor JOB [doris]
Posted by "morningman (via GitHub)" <gi...@apache.org>.
morningman merged PR #26845:
URL: https://github.com/apache/doris/pull/26845
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [WIP][Improve](Job)Refactor JOB [doris]
Posted by "CalvinKirs (via GitHub)" <gi...@apache.org>.
CalvinKirs commented on PR #26845:
URL: https://github.com/apache/doris/pull/26845#issuecomment-1808015473
UT failed has nothing to do with this change
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [WIP][Improve](Job)Refactor JOB [doris]
Posted by "doris-robot (via GitHub)" <gi...@apache.org>.
doris-robot commented on PR #26845:
URL: https://github.com/apache/doris/pull/26845#issuecomment-1808099504
(From new machine)TeamCity pipeline, clickbench performance test result:
the sum of best hot time: 45.53 seconds
stream load tsv: 553 seconds loaded 74807831229 Bytes, about 129 MB/s
stream load json: 20 seconds loaded 2358488459 Bytes, about 112 MB/s
stream load orc: 65 seconds loaded 1101869774 Bytes, about 16 MB/s
stream load parquet: 32 seconds loaded 861443392 Bytes, about 25 MB/s
insert into select: 29.1 seconds inserted 10000000 Rows, about 343K ops/s
storage size: 17162250464 Bytes
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [WIP][Improve](Job)Refactor JOB [doris]
Posted by "CalvinKirs (via GitHub)" <gi...@apache.org>.
CalvinKirs commented on PR #26845:
URL: https://github.com/apache/doris/pull/26845#issuecomment-1808479874
run buildall
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [Improve](Job)Refactor JOB [doris]
Posted by "CalvinKirs (via GitHub)" <gi...@apache.org>.
CalvinKirs commented on code in PR #26845:
URL: https://github.com/apache/doris/pull/26845#discussion_r1392144386
##########
fe/fe-core/src/main/java/org/apache/doris/job/executor/DefaultTaskExecutorHandler.java:
##########
@@ -0,0 +1,75 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.job.executor;
+
+import org.apache.doris.job.disruptor.ExecuteTaskEvent;
+import org.apache.doris.job.manager.TaskTokenManager;
+import org.apache.doris.job.task.AbstractTask;
+
+import com.lmax.disruptor.WorkHandler;
+import lombok.extern.slf4j.Slf4j;
+
+import java.util.concurrent.Semaphore;
+
+/**
+ * DefaultTaskExecutor is an implementation of the TaskExecutor interface.
+ * if you need to implement your own TaskExecutor, you could refer to this class. and need to register
+ * it in the TaskExecutorFactory
+ * It executes a given AbstractTask by acquiring a semaphore token from the TaskTokenManager
+ * and releasing it after the task execution.
+ */
+@Slf4j
+public class DefaultTaskExecutorHandler<T extends AbstractTask> implements WorkHandler<ExecuteTaskEvent<T>> {
+
+
+ @Override
+ public void onEvent(ExecuteTaskEvent<T> executeTaskEvent) {
+ T task = executeTaskEvent.getTask();
+ if (null == task) {
+ log.warn("task is null, ignore,maybe task has been canceled");
+ return;
+ }
+ if (task.isCancelled()) {
+ log.info("task is canceled, ignore");
+ return;
+ }
+ if (null == executeTaskEvent.getJobConfig().getMaxConcurrentTaskNum()
+ || executeTaskEvent.getJobConfig().getMaxConcurrentTaskNum() <= 0) {
+ try {
+ task.runTask();
+ return;
+ } catch (Exception e) {
+ log.warn("execute task error, task id is {}", task.getTaskId(), e);
Review Comment:
if task.onFail() throw exception, we will catch it here
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [Improve](Job)Refactor JOB [doris]
Posted by "CalvinKirs (via GitHub)" <gi...@apache.org>.
CalvinKirs commented on PR #26845:
URL: https://github.com/apache/doris/pull/26845#issuecomment-1809714119
run buildall
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org