dataroaring commented on code in PR #56175:
URL: https://github.com/apache/doris/pull/56175#discussion_r2357646376
##########
fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/InsertJob.java:
##########
@@ -96,9 +98,15 @@ public class InsertJob extends AbstractJob<InsertTask,
Map<Object, Object>> impl
.add(new Column("CreateTime", ScalarType.createStringType()))
.addAll(COMMON_SCHEMA)
.add(new Column("Comment", ScalarType.createStringType()))
+ // only execute type = streaming need record
+ .add(new Column("Properties", ScalarType.createStringType()))
+ .add(new Column("Progress", ScalarType.createStringType()))
+ .add(new Column("RemoteOffset", ScalarType.createStringType()))
Review Comment:
Remote stands for what/
##########
fe/fe-core/src/main/java/org/apache/doris/load/loadv2/LoadStatistic.java:
##########
@@ -123,6 +123,14 @@ public Map<String, String> getCounters() {
return counters;
}
+ public int getFileNumber() {
+ return fileNum;
+ }
+
+ public long getTotalFileSizeB() {
Review Comment:
Where is the function above called.
##########
fe/fe-core/src/main/java/org/apache/doris/job/scheduler/StreamingTaskScheduler.java:
##########
@@ -0,0 +1,118 @@
+// 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.catalog.Env;
+import org.apache.doris.common.Config;
+import org.apache.doris.common.CustomThreadFactory;
+import org.apache.doris.common.util.MasterDaemon;
+import org.apache.doris.job.exception.JobException;
+import org.apache.doris.job.extensions.insert.streaming.StreamingInsertJob;
+import org.apache.doris.job.extensions.insert.streaming.StreamingInsertTask;
+
+import lombok.extern.log4j.Log4j2;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.LinkedBlockingDeque;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+
+@Log4j2
+public class StreamingTaskScheduler extends MasterDaemon {
+ private final ThreadPoolExecutor threadPool = new ThreadPoolExecutor(
+ 0,
+ Config.max_streaming_job_num,
+ 60,
+ TimeUnit.SECONDS,
+ new ArrayBlockingQueue<>(Config.max_streaming_job_num),
+ new CustomThreadFactory("streaming-task-execute"),
+ new ThreadPoolExecutor.AbortPolicy()
+ );
+ private final ScheduledThreadPoolExecutor delayScheduler
+ = new ScheduledThreadPoolExecutor(1, new
CustomThreadFactory("streaming-task-delay-scheduler"));
+
+ public StreamingTaskScheduler() {
+ super("Streaming-task-scheduler", 1);
+ }
+
+ @Override
+ protected void runAfterCatalogReady() {
+ try {
+ process();
+ } catch (Throwable e) {
+ log.warn("Failed to process one round of StreamingTaskScheduler",
e);
+ }
+ }
+
+ private void process() throws InterruptedException {
+ List<StreamingInsertTask> tasks = new ArrayList<>();
+ LinkedBlockingDeque<StreamingInsertTask> needScheduleTasksQueue =
+
Env.getCurrentEnv().getJobManager().getStreamingTaskManager().getNeedScheduleTasksQueue();
+ tasks.add(needScheduleTasksQueue.take());
+ needScheduleTasksQueue.drainTo(tasks);
+ scheduleTasks(tasks);
+ }
+
+ private void scheduleTasks(List<StreamingInsertTask> tasks) {
+ for (StreamingInsertTask task : tasks) {
+ threadPool.execute(() -> {
+ try {
+ scheduleOneTask(task);
+ } catch (Exception e) {
+ log.error("Failed to schedule task, task id: {}, job id:
{}", task.getTaskId(), task.getJobId(), e);
Review Comment:
Will the task be scheduled later?
##########
fe/fe-core/src/main/java/org/apache/doris/fs/remote/RemoteFile.java:
##########
@@ -75,6 +77,22 @@ public void setPath(Path path) {
this.path = path;
}
+ public String getBucket() {
+ return bucket;
+ }
+
+ public void setBucket(String bucket) {
+ this.bucket = bucket;
+ }
+
+ public String getParentPath() {
+ return parentPath;
+ }
+
+ public void setParentPath(String parentPath) {
+ this.parentPath = parentPath;
Review Comment:
I can not find where is the function called.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
