[improvement](streaming) Support one-time S3 streaming ingestion

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

@@ -404,7 +404,7 @@ private String validateComputeGroupProperty(Map<String, String> props) throws An
     private SourceOffsetProvider createOffsetProvider(Map<String, String> jdbcSourceProps) {
         SourceOffsetProvider provider;
         if (tvfType != null) {
-            provider = SourceOffsetProviderFactory.createSourceOffsetProvider(tvfType);
+            provider = SourceOffsetProviderFactory.createSourceOffsetProvider(tvfType, jobProperties);
         } else {
             provider = new JdbcSourceOffsetProvider(getJobId(), dataSourceType, jdbcSourceProps);
         }
@@ -555,6 +555,9 @@ public void alterJob(AlterJobCommand alterJobCommand) throws AnalysisException,
     public void updateJobStatus(JobStatus status) throws JobException {
         lock.writeLock().lock();
         try {
+            if (isFinalStatus() && !getJobStatus().equals(status)) {
+                throw new JobException("Can't update final job status " + getJobStatus() + " to " + status);
+            }
             super.updateJobStatus(status);
             if (JobStatus.PAUSED.equals(getJobStatus())) {
                 clearRunningStreamTask(status);
@@ -592,6 +595,41 @@ public boolean updateJobStatusIfCurrent(JobStatus expectedStatus, JobStatus newS
         }
     }
 
+    public boolean tryFinishJob() throws JobException {
+        lock.writeLock().lock();
+        try {
+            if (!isActive()) {
+                return false;
+            }
+            if (runningStreamTask != null && TaskStatus.PENDING.equals(runningStreamTask.getStatus())) {
+                // Cancel the waiting task when a metadata scan detects the end of the source.
+                cancelAllTasks(false);
+            }
+            resetFailureInfo(null);
+            updateJobStatus(JobStatus.FINISHED);
+            logUpdateOperation();
+            return true;
+        } finally {
+            lock.writeLock().unlock();
+        }
+    }
+
+    private boolean tryPauseJob(FailureReason reason) throws JobException {
+        lock.writeLock().lock();
+        try {
+            if (!isActive()
+                    || (getFailureReason() != null
+                    && InternalErrorCode.MANUAL_PAUSE_ERR.equals(getFailureReason().getCode()))) {
+                return false;
+            }
+            updateJobStatus(JobStatus.PAUSED);
+            setFailureReason(reason);
+            return true;
+        } finally {
+            lock.writeLock().unlock();
+        }
+    }
+
     public void resetFailureInfo(FailureReason reason) {
         this.setFailureReason(reason);
         // Currently, only delayMsg is present here, which needs to be cleared when the status changes.
@@ -766,17 +804,8 @@ protected void fetchMeta() throws JobException {
             offsetProvider.fetchRemoteMeta(props);
         } catch (Exception ex) {
             log.warn("fetch remote meta failed, job id: {}", getJobId(), ex);
-            if (this.getFailureReason() == null
-                    || !InternalErrorCode.MANUAL_PAUSE_ERR.equals(this.getFailureReason().getCode())) {
-                // When a job is manually paused, it does not need to be set again,
-                // otherwise, it may be woken up by auto resume.
-                // Pause before setting the reason: updateJobStatus's writeLock orders this after any
-                // task-success callback that clears failureReason, so a success can't wipe the reason.
-                this.updateJobStatus(JobStatus.PAUSED);
-                this.setFailureReason(
-                        new FailureReason(InternalErrorCode.GET_REMOTE_DATA_ERROR,
-                                "Failed to fetch meta, " + ex.getMessage()));
-
+            if (tryPauseJob(new FailureReason(InternalErrorCode.GET_REMOTE_DATA_ERROR,
+                    "Failed to fetch meta, " + ex.getMessage()))) {
                 if (MetricRepo.isInit) {
                     MetricRepo.COUNTER_STREAMING_JOB_GET_META_FAIL_COUNT.increase(1L);
                 }
@@ -814,26 +843,24 @@ public void advanceSplitsIfNeed() throws JobException {
             }
         } catch (Exception ex) {
             log.warn("advance splits failed, job id: {}", getJobId(), ex);
-            if (this.getFailureReason() == null
-                    || !InternalErrorCode.MANUAL_PAUSE_ERR.equals(this.getFailureReason().getCode())) {
-                this.setFailureReason(new FailureReason(
-                        InternalErrorCode.GET_REMOTE_DATA_ERROR,
-                        "Failed to advance splits, " + ex.getMessage()));
-                this.updateJobStatus(JobStatus.PAUSED);
-            }
+            tryPauseJob(new FailureReason(InternalErrorCode.GET_REMOTE_DATA_ERROR,
+                    "Failed to advance splits, " + ex.getMessage()));
         }
     }
 
     public boolean needScheduleTask() {
         readLock();
         try {
-            return (getJobStatus().equals(JobStatus.RUNNING)
-                    || getJobStatus().equals(JobStatus.PENDING));
+            return isActive();
         } finally {
             readUnlock();
         }
     }
 
+    private boolean isActive() {
+        return JobStatus.PENDING.equals(getJobStatus()) || JobStatus.RUNNING.equals(getJobStatus());
+    }
+
     public void clearRunningStreamTask(JobStatus newJobStatus) {
         if (runningStreamTask != null) {
             log.info("clear running streaming insert task for job {}, task {}, status {} ",
@@ -972,6 +999,7 @@ private void updateCloudJobStatisticAndOffset(StreamingTaskTxnCommitAttachment a
         this.jobStatistic.setFileSize(attachment.getFileBytes());
         this.jobStatistic.setFilteredRows(attachment.getFilteredRows());
         offsetProvider.updateOffset(offsetProvider.deserializeOffset(attachment.getOffset()));
+        this.offsetProviderPersist = offsetProvider.getPersistInfo();
 
         //update metric
         if (MetricRepo.isInit && !isReplay) {
@@ -1482,16 +1510,16 @@ public void replayOnVisible(TransactionState txnState) {
 
     @Override
     public void gsonPostProcess() throws IOException {
+        if (jobProperties == null && properties != null) {
+            jobProperties = new StreamingJobProperties(properties);
+        }
         if (offsetProvider == null) {
             offsetProvider = createOffsetProvider(sourceProperties);
             if (tvfType != null) {
                 offsetProvider.restoreFromPersistInfo(offsetProviderPersist);
             }
         }
 
-        if (jobProperties == null && properties != null) {
-            jobProperties = new StreamingJobProperties(properties);
-        }
         recomputeDerivedFields();
 
         if (null == getSucceedTaskCount()) {

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

@@ -33,6 +33,7 @@
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
+import java.util.Locale;
 import java.util.Map;
 import java.util.Set;
 
@@ -41,12 +42,16 @@ public class StreamingJobProperties implements JobProperties {
     public static final String MAX_INTERVAL_SECOND_PROPERTY = "max_interval";
     public static final String S3_MAX_BATCH_FILES_PROPERTY = "s3.max_batch_files";
     public static final String S3_MAX_BATCH_BYTES_PROPERTY = "s3.max_batch_bytes";
+    public static final String S3_INGESTION_MODE_PROPERTY = "s3.ingestion_mode";
+    public static final String S3_INGESTION_MODE_LEXICAL = "LEXICAL";
+    public static final String S3_INGESTION_MODE_ONCE = "ONCE";
     public static final String SESSION_VAR_PREFIX = "session.";
     public static final String INTERNAL_KEY_PREFIX = "__";
     public static final String OFFSET_PROPERTY = "offset";
     public static final String COMPUTE_GROUP_PROPERTY = "compute_group";
     public static final List<String> SUPPORT_STREAM_JOB_PROPS = Arrays.asList(MAX_INTERVAL_SECOND_PROPERTY,
-            S3_MAX_BATCH_FILES_PROPERTY, S3_MAX_BATCH_BYTES_PROPERTY, OFFSET_PROPERTY, COMPUTE_GROUP_PROPERTY);
+            S3_MAX_BATCH_FILES_PROPERTY, S3_MAX_BATCH_BYTES_PROPERTY, S3_INGESTION_MODE_PROPERTY,
+            OFFSET_PROPERTY, COMPUTE_GROUP_PROPERTY);
 
     public static final long DEFAULT_MAX_INTERVAL_SECOND = 10;
     public static final long DEFAULT_MAX_S3_BATCH_FILES = 256;
@@ -119,6 +124,15 @@ public void validate() throws AnalysisException {
                         && v <= (long) (1024 * 1024 * 1024) * 10,
                 StreamingJobProperties.S3_MAX_BATCH_BYTES_PROPERTY + " should between 100MB and 10GB");
 
+        String ingestionMode = getS3IngestionMode();
+        if (!S3_INGESTION_MODE_LEXICAL.equals(ingestionMode)
+                && !S3_INGESTION_MODE_ONCE.equals(ingestionMode)) {
+            throw new AnalysisException("Unsupported s3.ingestion_mode: " + ingestionMode);
+        }
+        if (S3_INGESTION_MODE_ONCE.equals(ingestionMode) && properties.containsKey(OFFSET_PROPERTY)) {
+            throw new AnalysisException("offset is not supported when s3.ingestion_mode is ONCE");
+        }
+
         // validate session variables
         try {
             Map<String, String> sessionVarMap = parseSessionVarMap();
@@ -200,4 +214,13 @@ public String getOffsetProperty() {
     public String getComputeGroup() {
         return properties.get(COMPUTE_GROUP_PROPERTY);
     }
+
+    public String getS3IngestionMode() {
+        return properties.getOrDefault(S3_INGESTION_MODE_PROPERTY, S3_INGESTION_MODE_LEXICAL)
+                .trim().toUpperCase(Locale.ROOT);
+    }
+
+    public boolean isS3OnceMode() {
+        return S3_INGESTION_MODE_ONCE.equals(getS3IngestionMode());
+    }
 }

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

@@ -74,9 +74,7 @@ private void handlePendingState() throws JobException {
         }
         if (streamingInsertJob.hasReachedEnd()) {
             // Source already fully consumed (e.g. snapshot-only mode recovered after FE restart).
-            // Transition directly to FINISHED without creating a new task.
-            streamingInsertJob.updateJobStatus(JobStatus.FINISHED);
-            streamingInsertJob.logUpdateOperation();
+            streamingInsertJob.tryFinishJob();
             return;
         }
         streamingInsertJob.createStreamingTask();

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

@@ -18,6 +18,7 @@
 package org.apache.doris.job.offset;
 
 import org.apache.doris.job.exception.JobException;
+import org.apache.doris.job.extensions.insert.streaming.StreamingJobProperties;
 import org.apache.doris.job.offset.jdbc.JdbcTvfSourceOffsetProvider;
 import org.apache.doris.job.offset.s3.S3SourceOffsetProvider;
 
@@ -35,8 +36,12 @@ public class SourceOffsetProviderFactory {
         map.put("cdc_stream", JdbcTvfSourceOffsetProvider.class);
     }
 
-    public static SourceOffsetProvider createSourceOffsetProvider(String sourceType) {
+    public static SourceOffsetProvider createSourceOffsetProvider(
+            String sourceType, StreamingJobProperties jobProperties) {
         try {
+            if ("s3".equalsIgnoreCase(sourceType) && jobProperties.isS3OnceMode()) {
+                return new S3SourceOffsetProvider(jobProperties);
+            }
             Class<? extends SourceOffsetProvider> cla = map.get(sourceType.toLowerCase());
             if (cla == null) {
                 throw new JobException("Unsupported source type: " + sourceType);

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

@@ -34,6 +34,8 @@ public class S3Offset implements Offset {
     // s3://bucket/path/{1.csv,2.csv}
     String fileLists;
     int fileNum;
+    @SerializedName("lastBatch")
+    boolean lastBatch;
 
     @Override
     public String toSerializedJson() {

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

@@ -49,8 +49,18 @@
 
 @Log4j2
 public class S3SourceOffsetProvider implements SourceOffsetProvider {
-    S3Offset currentOffset;
-    String maxEndFile;
+    private final boolean onceMode;
+    private volatile S3Offset noMoreFilesAfterOffset;
+    volatile S3Offset currentOffset;
+    volatile String maxEndFile;
+
+    public S3SourceOffsetProvider() {
+        this.onceMode = false;
+    }
+
+    public S3SourceOffsetProvider(StreamingJobProperties jobProperties) {
+        this.onceMode = jobProperties.isS3OnceMode();
+    }
 
     @Override
     public String getSourceType() {
@@ -95,6 +105,7 @@ public S3Offset getNextOffset(StreamingJobProperties jobProps, Map<String, Strin
                 offset.setEndFile(lastFile);
                 offset.setFileNum(rfiles.size());
                 maxEndFile = globListing.getMaxFile();
+                offset.setLastBatch(onceMode && lastFile.equals(globListing.getMaxFile()));
             } else {
                 throw new RuntimeException("No new files found in path: " + filePath);
             }
@@ -159,7 +170,8 @@ public void fetchRemoteMeta(Map<String, String> properties) throws Exception {
         Map<String, String> copiedProps = Maps.newTreeMap(String.CASE_INSENSITIVE_ORDER);
         copiedProps.putAll(properties);
         StorageAdapter storageAdapter = StorageAdapter.of(copiedProps);
-        String startFile = currentOffset == null ? null : currentOffset.endFile;
+        S3Offset offsetAtScan = currentOffset;
+        String startFile = offsetAtScan == null ? null : offsetAtScan.endFile;
         try (FileSystem fileSystem = FileSystemFactory.getFileSystem(storageAdapter)) {
             String uri = storageAdapter.validateAndGetUri(copiedProps);
             String filePath = storageAdapter.validateAndNormalizeUri(uri);
@@ -168,6 +180,9 @@ public void fetchRemoteMeta(Map<String, String> properties) throws Exception {
                 throw new java.io.IOException("debug point: simulated S3 auth error");
             }
             GlobListing globListing = fileSystem.globListWithLimit(Location.of(filePath), startFile, 1, 1);
+            if (onceMode) {
+                noMoreFilesAfterOffset = globListing.getFiles().isEmpty() ? offsetAtScan : null;
+            }
             if (!globListing.getFiles().isEmpty() && StringUtils.isNotEmpty(globListing.getMaxFile())) {
                 maxEndFile = globListing.getMaxFile();
             }
@@ -176,6 +191,9 @@ public void fetchRemoteMeta(Map<String, String> properties) throws Exception {
 
     @Override
     public boolean hasMoreDataToConsume() {
+        if (hasReachedEnd()) {
+            return false;
+        }
         if (currentOffset == null || currentOffset.endFile == null) {
             return true;
         }
@@ -186,6 +204,12 @@ public boolean hasMoreDataToConsume() {
         return false;
     }
 
+    @Override
+    public boolean hasReachedEnd() {
+        S3Offset offset = currentOffset;
+        return onceMode && offset != null && (offset.isLastBatch() || noMoreFilesAfterOffset == offset);
+    }
+
     @Override
     public String getPersistInfo() {
         if (currentOffset == null) {

fe/fe-core/src/main/java/org/apache/doris/job/scheduler/StreamingTaskScheduler.java

@@ -104,7 +104,7 @@ private void scheduleTasks(List<AbstractStreamingTask> tasks) {
         }
     }
 
-    private void scheduleOneTask(AbstractStreamingTask task) {
+    private void scheduleOneTask(AbstractStreamingTask task) throws JobException {
         if (DebugPointUtil.isEnable("StreamingJob.scheduleTask.exception")) {
             throw new RuntimeException("debug point StreamingJob.scheduleTask.exception");
         }
@@ -122,6 +122,10 @@ private void scheduleOneTask(AbstractStreamingTask task) {
         }
         // reject task if no more data to consume
         if (!job.hasMoreDataToConsume()) {
+            if (job.hasReachedEnd()) {
+                job.tryFinishJob();
+                return;
+            }
             String delayMsg = "No data available for consumption at the moment, will retry after "
                     + (System.currentTimeMillis() + DELAY_SCHEDULER_MS);
             job.setJobRuntimeMsg(delayMsg);

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AlterJobCommand.java

@@ -41,6 +41,7 @@
 import com.google.common.base.Preconditions;
 import org.apache.commons.lang3.StringUtils;
 
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
@@ -277,11 +278,19 @@ private void checkUnmodifiableSourceProperties(Map<String, String> originSourceP
     }
 
     private void validateProps(StreamingInsertJob streamingJob) throws AnalysisException {
-        StreamingJobProperties jobProperties = new StreamingJobProperties(properties);
-        jobProperties.validate();
-        if (jobProperties.getOffsetProperty() != null) {
-            streamingJob.validateAlterOffset(jobProperties.getOffsetProperty());
-            streamingJob.validateOffset(jobProperties.getOffsetProperty());
+        StreamingJobProperties originJobProperties =
+                new StreamingJobProperties(streamingJob.getProperties());
+        Map<String, String> mergedProperties = new HashMap<>(streamingJob.getProperties());
+        mergedProperties.putAll(properties);
+        StreamingJobProperties updatedJobProperties = new StreamingJobProperties(mergedProperties);
+        updatedJobProperties.validate();
+        if (!originJobProperties.getS3IngestionMode().equals(updatedJobProperties.getS3IngestionMode())) {
+            throw new AnalysisException("s3.ingestion_mode cannot be altered");
+        }
+        String offset = properties.get(StreamingJobProperties.OFFSET_PROPERTY);
+        if (offset != null) {
+            streamingJob.validateAlterOffset(offset);
+            streamingJob.validateOffset(offset);
         }
     }