[Feature](WIP) Add Create StreamingJob and Alter Job

fe/fe-core/src/main/antlr4/org/apache/doris/nereids/DorisParser.g4

@@ -99,16 +99,19 @@ materializedViewStatement
     | SHOW CREATE MATERIALIZED VIEW mvName=multipartIdentifier                                  #showCreateMTMV
     ;
 supportedJobStatement
-    : CREATE JOB label=multipartIdentifier ON SCHEDULE
-        (
+    : CREATE JOB label=multipartIdentifier propertyClause?
+      ON (STREAMING | SCHEDULE(
             (EVERY timeInterval=INTEGER_VALUE timeUnit=identifier
             (STARTS (startTime=STRING_LITERAL | CURRENT_TIMESTAMP))?
             (ENDS endsTime=STRING_LITERAL)?)
             |
-            (AT (atTime=STRING_LITERAL | CURRENT_TIMESTAMP)))
-        commentSpec?
-        DO supportedDmlStatement                                                                                                             #createScheduledJob
+            (AT (atTime=STRING_LITERAL | CURRENT_TIMESTAMP))
+            )
+         )
+       commentSpec?
+       DO supportedDmlStatement                                                                                                             #createScheduledJob
    | PAUSE JOB WHERE (jobNameKey=identifier) EQ (jobNameValue=STRING_LITERAL)                                                                #pauseJob
+   | ALTER JOB FOR (jobNameKey=identifier) (propertyClause | supportedDmlStatement | propertyClause  supportedDmlStatement)                  #alterJob
    | DROP JOB (IF EXISTS)? WHERE (jobNameKey=identifier) EQ (jobNameValue=STRING_LITERAL)                                                    #dropJob
    | RESUME JOB WHERE (jobNameKey=identifier) EQ (jobNameValue=STRING_LITERAL)                                                               #resumeJob
    | CANCEL TASK WHERE (jobNameKey=identifier) EQ (jobNameValue=STRING_LITERAL) AND (taskIdKey=identifier) EQ (taskIdValue=INTEGER_VALUE)    #cancelJobTask

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

@@ -0,0 +1,25 @@
+// 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.common.AnalysisException;
+
+public interface JobProperties {
+    default void validate() throws AnalysisException {
+    }
+}

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

@@ -51,6 +51,8 @@
 import org.apache.doris.qe.ConnectContext;
 import org.apache.doris.qe.ShowResultSetMetaData;
 import org.apache.doris.qe.StmtExecutor;
+import org.apache.doris.thrift.TCell;
+import org.apache.doris.thrift.TRow;
 import org.apache.doris.thrift.TUniqueId;
 import org.apache.doris.transaction.ErrorTabletInfo;
 import org.apache.doris.transaction.TabletCommitInfo;
@@ -96,6 +98,11 @@ 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("Progress", ScalarType.createStringType()))
+            .add(new Column("RemoteOffset", ScalarType.createStringType()))
+            .add(new Column("LoadStatistic", ScalarType.createStringType()))
+            .add(new Column("ErrorMsg", ScalarType.createStringType()))
             .build();
 
     private static final ShowResultSetMetaData TASK_META_DATA =
@@ -112,6 +119,8 @@ public class InsertJob extends AbstractJob<InsertTask, Map<Object, Object>> impl
                     .addColumn(new Column("TrackingUrl", ScalarType.createVarchar(200)))
                     .addColumn(new Column("LoadStatistic", ScalarType.createVarchar(200)))
                     .addColumn(new Column("User", ScalarType.createVarchar(50)))
+                    // only execute type = streaming need record
+                    .addColumn(new Column("Offset", ScalarType.createStringType()))
                     .build();
 
     public static final ImmutableMap<String, Integer> COLUMN_TO_INDEX;
@@ -523,6 +532,28 @@ public List<String> getShowInfo() {
         }
     }
 
+    @Override
+    public TRow getTvfInfo() {
+        TRow trow = new TRow();
+        trow.addToColumnValue(new TCell().setStringVal(String.valueOf(getJobId())));
+        trow.addToColumnValue(new TCell().setStringVal(getJobName()));
+        trow.addToColumnValue(new TCell().setStringVal(getCreateUser().getQualifiedUser()));
+        trow.addToColumnValue(new TCell().setStringVal(getJobConfig().getExecuteType().name()));
+        trow.addToColumnValue(new TCell().setStringVal(getJobConfig().convertRecurringStrategyToString()));
+        trow.addToColumnValue(new TCell().setStringVal(getJobStatus().name()));
+        trow.addToColumnValue(new TCell().setStringVal(getExecuteSql()));
+        trow.addToColumnValue(new TCell().setStringVal(TimeUtils.longToTimeString(getCreateTimeMs())));
+        trow.addToColumnValue(new TCell().setStringVal(String.valueOf(getSucceedTaskCount().get())));
+        trow.addToColumnValue(new TCell().setStringVal(String.valueOf(getFailedTaskCount().get())));
+        trow.addToColumnValue(new TCell().setStringVal(String.valueOf(getCanceledTaskCount().get())));
+        trow.addToColumnValue(new TCell().setStringVal(getComment()));
+        trow.addToColumnValue(new TCell().setStringVal(FeConstants.null_string));
+        trow.addToColumnValue(new TCell().setStringVal(FeConstants.null_string));
+        trow.addToColumnValue(new TCell().setStringVal(loadStatistic.toJson()));
+        trow.addToColumnValue(new TCell().setStringVal(failMsg == null ? FeConstants.null_string : failMsg.getMsg()));
+        return trow;
+    }
+
     @Override
     public String formatMsgWhenExecuteQueueFull(Long taskId) {
         return commonFormatMsgWhenExecuteQueueFull(taskId, "insert_task_queue_size",

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

@@ -67,7 +67,8 @@ public class InsertTask extends AbstractTask {
             new Column("FinishTime", ScalarType.createStringType()),
             new Column("TrackingUrl", ScalarType.createStringType()),
             new Column("LoadStatistic", ScalarType.createStringType()),
-            new Column("User", ScalarType.createStringType()));
+            new Column("User", ScalarType.createStringType()),
+            new Column("Offset", ScalarType.createStringType()));
 
     public static final ImmutableMap<String, Integer> COLUMN_TO_INDEX;
 
@@ -272,6 +273,7 @@ public TRow getTvfInfo(String jobName) {
         } else {
             trow.addToColumnValue(new TCell().setStringVal(userIdentity.getQualifiedUser()));
         }
+        trow.addToColumnValue(new TCell().setStringVal(""));
         return trow;
     }
 
@@ -292,6 +294,7 @@ private TRow getPendingTaskTVFInfo(String jobName) {
         trow.addToColumnValue(new TCell().setStringVal(""));
         trow.addToColumnValue(new TCell().setStringVal(""));
         trow.addToColumnValue(new TCell().setStringVal(userIdentity.getQualifiedUser()));
+        trow.addToColumnValue(new TCell().setStringVal(""));
         return trow;
     }
 

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

@@ -17,13 +17,16 @@
 
 package org.apache.doris.job.extensions.insert.streaming;
 
+import org.apache.doris.analysis.UserIdentity;
 import org.apache.doris.common.io.Text;
 import org.apache.doris.job.base.AbstractJob;
+import org.apache.doris.job.base.JobExecutionConfiguration;
 import org.apache.doris.job.common.JobStatus;
 import org.apache.doris.job.common.JobType;
 import org.apache.doris.job.common.PauseReason;
 import org.apache.doris.job.exception.JobException;
 import org.apache.doris.persist.gson.GsonUtils;
+import org.apache.doris.qe.ConnectContext;
 
 import com.google.gson.annotations.SerializedName;
 import lombok.Getter;
@@ -37,6 +40,8 @@
 
 public class StreamingInsertJob extends AbstractJob<StreamingJobSchedulerTask, Map<Object, Object>> {
 
+    @SerializedName("did")
+    private final long dbId;
     @Getter
     @SerializedName("st")
     protected JobStatus status;
@@ -52,6 +57,26 @@ public class StreamingInsertJob extends AbstractJob<StreamingJobSchedulerTask, M
     @Setter
     protected long autoResumeCount;
 
+    @Getter
+    @SerializedName("jp")
+    private StreamingJobProperties jobProperties;
+
+    public StreamingInsertJob(String jobName,
+            JobStatus jobStatus,
+            String dbName,
+            String comment,
+            UserIdentity createUser,
+            JobExecutionConfiguration jobConfig,
+            Long createTimeMs,
+            String executeSql,
+            StreamingJobProperties jobProperties) {
+        super(getNextJobId(), jobName, jobStatus, dbName, comment, createUser,
+                jobConfig, createTimeMs, executeSql);
+        this.dbId = ConnectContext.get().getCurrentDbId();
+        this.jobProperties = jobProperties;
+    }
+
+
     @Override
     public void updateJobStatus(JobStatus status) throws JobException {
         super.updateJobStatus(status);

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

@@ -0,0 +1,64 @@
+// 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.streaming;
+
+import org.apache.doris.common.AnalysisException;
+import org.apache.doris.common.util.Util;
+import org.apache.doris.job.base.JobProperties;
+
+import lombok.Data;
+
+import java.util.Map;
+
+@Data
+public class StreamingJobProperties implements JobProperties {
+    public static final String MAX_INTERVAL_SECOND_PROPERTY = "max_interval";
+    public static final String S3_BATCH_FILES_PROPERTY = "s3.batch_files";
+    public static final String S3_BATCH_SIZE_PROPERTY = "s3.batch_size";
+    public static final long DEFAULT_MAX_INTERVAL_SECOND = 10;
+    public static final long DEFAULT_S3_BATCH_FILES = 256;
+    public static final long DEFAULT_S3_BATCH_SIZE = 10 * 1024 * 1024 * 1024L; // 10GB
+    public static final long DEFAULT_INSERT_TIMEOUT = 30 * 60; // 30min
+
+    private final Map<String, String> properties;
+    private long maxIntervalSecond;
+    private long s3BatchFiles;
+    private long s3BatchSize;
+
+    public StreamingJobProperties(Map<String, String> jobProperties) {
+        this.properties = jobProperties;
+    }
+
+    @Override
+    public void validate() throws AnalysisException {
+        this.maxIntervalSecond = Util.getLongPropertyOrDefault(
+                properties.get(StreamingJobProperties.MAX_INTERVAL_SECOND_PROPERTY),
+                StreamingJobProperties.DEFAULT_MAX_INTERVAL_SECOND, (v) -> v >= 1,
+                StreamingJobProperties.MAX_INTERVAL_SECOND_PROPERTY + " should > 1");
+
+        this.s3BatchFiles = Util.getLongPropertyOrDefault(
+                properties.get(StreamingJobProperties.S3_BATCH_FILES_PROPERTY),
+                StreamingJobProperties.DEFAULT_S3_BATCH_FILES, (v) -> v >= 1,
+                StreamingJobProperties.S3_BATCH_FILES_PROPERTY + " should >=1 ");
+
+        this.s3BatchSize = Util.getLongPropertyOrDefault(properties.get(StreamingJobProperties.S3_BATCH_SIZE_PROPERTY),
+                StreamingJobProperties.DEFAULT_S3_BATCH_SIZE, (v) -> v >= 100 * 1024 * 1024
+                        && v <= (long) (1024 * 1024 * 1024) * 10,
+                StreamingJobProperties.S3_BATCH_SIZE_PROPERTY + " should between 100MB and 10GB");
+    }
+}

fe/fe-core/src/main/java/org/apache/doris/job/manager/JobManager.java

@@ -32,6 +32,7 @@
 import org.apache.doris.common.util.LogKey;
 import org.apache.doris.datasource.InternalCatalog;
 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.JobType;
 import org.apache.doris.job.common.TaskType;
@@ -219,6 +220,17 @@ public void alterJobStatus(Long jobId, JobStatus status) throws JobException {
         jobMap.get(jobId).logUpdateOperation();
     }
 
+    public void alterJob(T job) {
+        writeLock();
+        try {
+            jobMap.put(job.getJobId(), job);
+            job.logUpdateOperation();
+        } finally {
+            writeUnlock();
+        }
+        log.info("update job success, jobId: {}", job.getJobId());
+    }
+
     public void alterJobStatus(String jobName, JobStatus jobStatus) throws JobException {
         for (T a : jobMap.values()) {
             if (a.getJobName().equals(jobName)) {
@@ -349,6 +361,9 @@ public void replayDeleteJob(T replayJob) throws JobException {
      */
     public void cancelTaskById(String jobName, Long taskId) throws JobException {
         for (T job : jobMap.values()) {
+            if (job.getJobConfig().getExecuteType().equals(JobExecuteType.STREAMING)) {
+                throw new JobException("streaming job not support cancel task by id");
+            }
             if (job.getJobName().equals(jobName)) {
                 job.cancelTaskById(taskId);
                 job.logUpdateOperation();
@@ -392,6 +407,14 @@ public T getJob(Long jobId) {
         return jobMap.get(jobId);
     }
 
+    public T getJobByName(String jobName) throws JobException {
+        for (T a : jobMap.values()) {
+            if (a.getJobName().equals(jobName)) {
+                return a;
+            }
+        }
+        throw new JobException("job not exist, jobName:" + jobName);
+    }
 
     /**
      * get load info by db

fe/fe-core/src/main/java/org/apache/doris/job/offset/Offset.java

@@ -0,0 +1,22 @@
+// 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.offset;
+
+public interface Offset {
+    String toJson();
+}

fe/fe-core/src/main/java/org/apache/doris/job/offset/SourceOffsetProvider.java

@@ -0,0 +1,62 @@
+// 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.offset;
+
+import org.apache.doris.nereids.trees.plans.commands.insert.InsertIntoTableCommand;
+
+/**
+ * Interface for managing offsets and metadata of a data source.
+ */
+public interface SourceOffsetProvider {
+    /**
+     * Get source type, e.g. s3, kafka
+     * @return
+     */
+    String getSourceType();
+
+    /**
+     * Get next offset to consume
+     * @return
+     */
+    Offset getNextOffset();
+
+    /**
+     * Rewrite the TVF parameters in the InsertIntoTableCommand based on the current offset.
+     * @param command
+     * @return
+     */
+    InsertIntoTableCommand rewriteTvfParamsInCommand(InsertIntoTableCommand command);
+
+    /**
+     * Update the progress of the source.
+     * @param offset
+     */
+    void updateProgress(Offset offset);
+
+    /**
+     * Fetch remote meta information, such as listing files in S3 or getting latest offsets in Kafka.
+     */
+    void fetchRemoteMeta();
+
+    /**
+     * Whether there is more data to consume
+     * @return
+     */
+    boolean hasMoreData();
+}
+