[Feature](streamjob) Streaming job support cdc_stream TVF (#61826)

### What problem does this PR solve?

Previously, streaming jobs only supported the non-TVF path (FROM Source
TO Database) for CDC ingestion. This PR adds support for streaming jobs
to drive CDC ingestion via the
cdc_stream Table-Valued Function (TVF), enabling a more flexible and
SQL-native approach.

```sql
CREATE JOB test_streaming_job_cdc_stream_postgres_name
            ON STREAMING DO INSERT INTO db.table 
            SELECT name, age FROM cdc_stream(
                "type"         = "postgres",
                "jdbc_url"     = "jdbc:postgresql://127.0.0.1:5432/postgres",
                "driver_url"   = "postgresql-42.5.0.jar",
                "driver_class" = "org.postgresql.Driver",
                "user"         = "postgres",
                "password"     = "123456",
                "database"     = "postgres",
                "schema"             = "cdc_test",
                "table"              = "test_streaming_job_cdc_stream_postgres_src",
                "offset"             = "initial"
            )                                                                                                                                                             
```

Core flow:
1. Snapshot phase: FE fetches split chunks from BE, persists them to the
meta table, then creates one StreamingInsertTask per split (or batch).
The TVF parameters are
rewritten per-task with meta (split boundary info), job.id, and task.id.
2. Binlog phase: After all snapshot splits are completed, transitions to
continuous binlog ingestion via the same cdc_stream TVF with binlog
offset as meta.
3. Offset commit (2PC safe): After fetchRecordStream completes on BE,
the actual end offset is stored in PipelineCoordinator.taskOffsetCache.
FE pulls this via brpc in
beforeCommitted and stores it in the txn attachment — ensuring the
offset is committed atomically with the data.
4. FE restart recovery: State is recovered via txn replay
(replayOnCommitted) rather than EditLog, rebuilding
chunkHighWatermarkMap and remainingSplits from the meta table.

Author: JNSimba

fe/fe-common/src/main/java/org/apache/doris/job/cdc/DataSourceConfigKeys.java

@@ -35,6 +35,7 @@ public class DataSourceConfigKeys {
     public static final String OFFSET_LATEST = "latest";
     public static final String OFFSET_SNAPSHOT = "snapshot";
     public static final String SNAPSHOT_SPLIT_SIZE = "snapshot_split_size";
+    public static final String SNAPSHOT_SPLIT_KEY = "snapshot_split_key";
     public static final String SNAPSHOT_PARALLELISM = "snapshot_parallelism";
     public static final String SNAPSHOT_PARALLELISM_DEFAULT = "1";
     public static final String SSL_MODE = "ssl_mode";

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

@@ -33,6 +33,8 @@
 import lombok.extern.log4j.Log4j2;
 import org.apache.commons.lang3.StringUtils;
 
+import java.util.Collections;
+import java.util.List;
 import java.util.concurrent.atomic.AtomicBoolean;
 
 @Log4j2
@@ -68,6 +70,15 @@ public AbstractStreamingTask(long jobId, long taskId, UserIdentity userIdentity)
 
     public abstract void run() throws JobException;
 
+    /**
+     * Returns the IDs of backends that ran the scan node for this task.
+     * Subclasses backed by a TVF query (e.g. StreamingInsertTask) override this
+     * to return the actual scan backend IDs from the coordinator.
+     */
+    public List<Long> getScanBackendIds() {
+        return Collections.emptyList();
+    }
+
     public abstract boolean onSuccess() throws JobException;
 
     public abstract void closeOrReleaseResources();

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

@@ -51,6 +51,7 @@
 import org.apache.doris.job.offset.SourceOffsetProvider;
 import org.apache.doris.job.offset.SourceOffsetProviderFactory;
 import org.apache.doris.job.offset.jdbc.JdbcSourceOffsetProvider;
+import org.apache.doris.job.offset.jdbc.JdbcTvfSourceOffsetProvider;
 import org.apache.doris.job.util.StreamingJobUtils;
 import org.apache.doris.load.loadv2.LoadJob;
 import org.apache.doris.load.loadv2.LoadStatistic;
@@ -72,6 +73,7 @@
 import org.apache.doris.qe.ShowResultSetMetaData;
 import org.apache.doris.rpc.RpcException;
 import org.apache.doris.service.FrontendOptions;
+import org.apache.doris.tablefunction.S3TableValuedFunction;
 import org.apache.doris.thrift.TCell;
 import org.apache.doris.thrift.TRow;
 import org.apache.doris.transaction.TransactionException;
@@ -300,8 +302,11 @@ private void initInsertJob() {
             this.tvfType = currentTvf.getFunctionName();
             this.originTvfProps = currentTvf.getProperties().getMap();
             this.offsetProvider = SourceOffsetProviderFactory.createSourceOffsetProvider(currentTvf.getFunctionName());
-            // validate offset props
-            if (jobProperties.getOffsetProperty() != null) {
+            this.offsetProvider.ensureInitialized(getJobId(), originTvfProps);
+            this.offsetProvider.initOnCreate();
+            // validate offset props, only for s3 cause s3 tvf no offset prop
+            if (jobProperties.getOffsetProperty() != null
+                    && S3TableValuedFunction.NAME.equalsIgnoreCase(tvfType)) {
                 Offset offset = validateOffset(jobProperties.getOffsetProperty());
                 this.offsetProvider.updateOffset(offset);
             }
@@ -536,6 +541,8 @@ protected void fetchMeta() throws JobException {
                 if (originTvfProps == null) {
                     this.originTvfProps = getCurrentTvf().getProperties().getMap();
                 }
+                // when fe restart, offsetProvider.jobId may be null
+                offsetProvider.ensureInitialized(getJobId(), originTvfProps);
                 offsetProvider.fetchRemoteMeta(originTvfProps);
             } else {
                 offsetProvider.fetchRemoteMeta(new HashMap<>());
@@ -653,7 +660,14 @@ private void updateJobStatisticAndOffset(StreamingTaskTxnCommitAttachment attach
         this.jobStatistic.setLoadBytes(this.jobStatistic.getLoadBytes() + attachment.getLoadBytes());
         this.jobStatistic.setFileNumber(this.jobStatistic.getFileNumber() + attachment.getNumFiles());
         this.jobStatistic.setFileSize(this.jobStatistic.getFileSize() + attachment.getFileBytes());
-        offsetProvider.updateOffset(offsetProvider.deserializeOffset(attachment.getOffset()));
+        Offset endOffset = offsetProvider.deserializeOffset(attachment.getOffset());
+        offsetProvider.updateOffset(endOffset);
+        if (!isReplay) {
+            offsetProvider.onTaskCommitted(attachment.getScannedRows(), attachment.getLoadBytes());
+            if (runningStreamTask != null) {
+                offsetProvider.applyEndOffsetToTask(runningStreamTask.getRunningOffset(), endOffset);
+            }
+        }
 
         //update metric
         if (MetricRepo.isInit && !isReplay) {
@@ -757,7 +771,8 @@ public void replayOnUpdated(StreamingInsertJob replayJob) {
      */
     private void modifyPropertiesInternal(Map<String, String> inputProperties) throws AnalysisException, JobException {
         StreamingJobProperties inputStreamProps = new StreamingJobProperties(inputProperties);
-        if (StringUtils.isNotEmpty(inputStreamProps.getOffsetProperty()) && this.tvfType != null) {
+        if (StringUtils.isNotEmpty(inputStreamProps.getOffsetProperty())
+                && S3TableValuedFunction.NAME.equalsIgnoreCase(this.tvfType)) {
             Offset offset = validateOffset(inputStreamProps.getOffsetProperty());
             this.offsetProvider.updateOffset(offset);
             if (Config.isCloudMode()) {
@@ -1016,14 +1031,25 @@ public void beforeCommitted(TransactionState txnState) throws TransactionExcepti
             }
             LoadJob loadJob = loadJobs.get(0);
             LoadStatistic loadStatistic = loadJob.getLoadStatistic();
+
+            String offsetJson = offsetProvider.getCommitOffsetJson(
+                    runningStreamTask.getRunningOffset(),
+                    runningStreamTask.getTaskId(),
+                    runningStreamTask.getScanBackendIds());
+
+
+            if (StringUtils.isBlank(offsetJson)) {
+                throw new TransactionException("Cannot find offset for attachment, load job id is "
+                        + runningStreamTask.getTaskId());
+            }
             txnState.setTxnCommitAttachment(new StreamingTaskTxnCommitAttachment(
                         getJobId(),
                         runningStreamTask.getTaskId(),
                         loadStatistic.getScannedRows(),
                         loadStatistic.getLoadBytes(),
                         loadStatistic.getFileNumber(),
                         loadStatistic.getTotalFileSizeB(),
-                        runningStreamTask.getRunningOffset().toSerializedJson()));
+                        offsetJson));
         } finally {
             if (shouldReleaseLock) {
                 writeUnlock();
@@ -1186,10 +1212,7 @@ public void commitOffset(CommitOffsetRequest offsetRequest) throws JobException
                 }
                 checkDataQuality(offsetRequest);
                 updateNoTxnJobStatisticAndOffset(offsetRequest);
-                if (offsetRequest.getScannedRows() == 0 && offsetRequest.getLoadBytes() == 0) {
-                    JdbcSourceOffsetProvider op = (JdbcSourceOffsetProvider) offsetProvider;
-                    op.setHasMoreData(false);
-                }
+                offsetProvider.onTaskCommitted(offsetRequest.getScannedRows(), offsetRequest.getLoadBytes());
                 if (offsetRequest.getTableSchemas() != null) {
                     JdbcSourceOffsetProvider op = (JdbcSourceOffsetProvider) offsetProvider;
                     op.setTableSchemas(offsetRequest.getTableSchemas());
@@ -1278,6 +1301,7 @@ public boolean hasReachedEnd() {
      */
     public void cleanup() throws JobException {
         log.info("cleanup streaming job {}", getJobId());
+
         // s3 tvf clean offset
         if (tvfType != null && Config.isCloudMode()) {
             Cloud.DeleteStreamingJobResponse resp = null;
@@ -1298,6 +1322,14 @@ public void cleanup() throws JobException {
             }
         }
 
+        // For TVF path, provider fields may be null after FE restart
+        if (this.offsetProvider instanceof JdbcTvfSourceOffsetProvider) {
+            if (originTvfProps == null) {
+                this.originTvfProps = getCurrentTvf().getProperties().getMap();
+            }
+            offsetProvider.ensureInitialized(getJobId(), originTvfProps);
+        }
+
         if (this.offsetProvider instanceof JdbcSourceOffsetProvider) {
             // jdbc clean chunk meta table
             ((JdbcSourceOffsetProvider) this.offsetProvider).cleanMeta(getJobId());

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

@@ -43,6 +43,7 @@
 import lombok.extern.log4j.Log4j2;
 
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
@@ -98,7 +99,7 @@ public void before() throws Exception {
         if (!baseCommand.getParsedPlan().isPresent()) {
             throw new JobException("Can not get Parsed plan");
         }
-        this.taskCommand = offsetProvider.rewriteTvfParams(baseCommand, runningOffset);
+        this.taskCommand = offsetProvider.rewriteTvfParams(baseCommand, runningOffset, getTaskId());
         this.taskCommand.setLabelName(Optional.of(labelName));
         this.stmtExecutor = new StmtExecutor(ctx, new LogicalPlanAdapter(taskCommand, ctx.getStatementContext()));
     }
@@ -126,6 +127,14 @@ public void run() throws JobException {
         }
     }
 
+    @Override
+    public List<Long> getScanBackendIds() {
+        if (stmtExecutor != null && stmtExecutor.getCoord() != null) {
+            return stmtExecutor.getCoord().getScanBackendIds();
+        }
+        return Collections.emptyList();
+    }
+
     @Override
     public boolean onSuccess() throws JobException {
         if (getIsCanceled().get()) {

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

@@ -192,6 +192,8 @@ private WriteRecordRequest buildRequestParams() throws JobException {
         Map<String, String> props = generateStreamLoadProps();
         request.setStreamLoadProps(props);
 
+        //`meta` refers to the data synchronized by the job in this instance,
+        // while `sourceProperties.offset` is the data entered by the user.
         Map<String, Object> splitMeta = offset.generateMeta();
         Preconditions.checkArgument(!splitMeta.isEmpty(), "split meta is empty");
         request.setMeta(splitMeta);

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

@@ -22,6 +22,7 @@
 import org.apache.doris.job.extensions.insert.streaming.StreamingJobProperties;
 import org.apache.doris.nereids.trees.plans.commands.insert.InsertIntoTableCommand;
 
+import java.util.List;
 import java.util.Map;
 
 /**
@@ -36,6 +37,20 @@ public interface SourceOffsetProvider {
      */
     String getSourceType();
 
+    /**
+     * Initialize the offset provider with job ID and original TVF properties.
+     * Only sets in-memory fields; safe to call on both fresh start and FE restart.
+     * May perform remote calls (e.g. fetching snapshot splits), so throws JobException.
+     */
+    default void ensureInitialized(Long jobId, Map<String, String> originTvfProps) throws JobException {}
+
+    /**
+     * Performs one-time initialization that must run only on fresh job creation, not on FE restart.
+     * For example, fetching and persisting snapshot splits to the meta table.
+     * Default: no-op (most providers need no extra setup).
+     */
+    default void initOnCreate() throws JobException {}
+
     /**
      * Get next offset to consume
      *
@@ -59,11 +74,12 @@ public interface SourceOffsetProvider {
 
     /**
      * Rewrite the TVF parameters in the SQL based on the current offset.
+     * Only implemented by TVF-based providers (e.g. S3, cdc_stream).
      *
      * @param nextOffset
      * @return rewritten InsertIntoTableCommand
      */
-    InsertIntoTableCommand rewriteTvfParams(InsertIntoTableCommand originCommand, Offset nextOffset);
+    InsertIntoTableCommand rewriteTvfParams(InsertIntoTableCommand originCommand, Offset nextOffset, long taskId);
 
     /**
      * Update the offset of the source.
@@ -110,6 +126,30 @@ default String getPersistInfo() {
         return null;
     }
 
+    /**
+     * Returns the serialized JSON offset to store in txn commit attachment.
+     * Default: serialize running offset directly (e.g. S3 path).
+     * CDC stream TVF overrides to pull actual end offset from BE after fetchRecordStream completes.
+     * scanBackendIds: IDs of the BEs that ran the TVF scan node, used to locate taskOffsetCache.
+     */
+    default String getCommitOffsetJson(Offset runningOffset, long taskId, List<Long> scanBackendIds) {
+        return runningOffset.toSerializedJson();
+    }
+
+    /**
+     * Called after each task is committed. Providers that track data availability
+     * (e.g. JDBC binlog) can use this to update internal state such as hasMoreData.
+     * Default: no-op.
+     */
+    default void onTaskCommitted(long scannedRows, long loadBytes) {}
+
+    /**
+     * Applies the end offset from a committed task back onto the running offset object
+     * in-place, so that showRange() can display the full [start, end] interval.
+     * Default: no-op (only meaningful for JDBC providers).
+     */
+    default void applyEndOffsetToTask(Offset runningOffset, Offset endOffset) {}
+
     /**
      * Returns true if the provider has reached a natural completion point
      * and the job should be marked as FINISHED.

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.offset.jdbc.JdbcTvfSourceOffsetProvider;
 import org.apache.doris.job.offset.s3.S3SourceOffsetProvider;
 
 import lombok.extern.log4j.Log4j2;
@@ -31,6 +32,7 @@ public class SourceOffsetProviderFactory {
 
     static {
         map.put("s3", S3SourceOffsetProvider.class);
+        map.put("cdc_stream", JdbcTvfSourceOffsetProvider.class);
     }
 
     public static SourceOffsetProvider createSourceOffsetProvider(String sourceType) {

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

@@ -70,10 +70,10 @@
 public class JdbcSourceOffsetProvider implements SourceOffsetProvider {
     public static final String SPLIT_ID = "splitId";
     private static final ObjectMapper objectMapper = new ObjectMapper();
-    private final int snapshotParallelism;
-    private Long jobId;
-    private DataSourceType sourceType;
-    private Map<String, String> sourceProperties = new HashMap<>();
+    protected int snapshotParallelism = 1;
+    protected Long jobId;
+    protected DataSourceType sourceType;
+    protected Map<String, String> sourceProperties = new HashMap<>();
 
     List<SnapshotSplit> remainingSplits = new ArrayList<>();
     List<SnapshotSplit> finishedSplits = new ArrayList<>();
@@ -92,6 +92,16 @@ public class JdbcSourceOffsetProvider implements SourceOffsetProvider {
 
     volatile boolean hasMoreData = true;
 
+    /**
+     * No-arg constructor for subclass use.
+     */
+    public JdbcSourceOffsetProvider() {
+        this.chunkHighWatermarkMap = new HashMap<>();
+    }
+
+    /**
+     * Constructor for FROM Source TO Database.
+     */
     public JdbcSourceOffsetProvider(Long jobId, DataSourceType sourceType, Map<String, String> sourceProperties) {
         this.jobId = jobId;
         this.sourceType = sourceType;
@@ -155,10 +165,13 @@ public String getShowMaxOffset() {
         return null;
     }
 
+    /**
+     * Should never call this.
+     */
     @Override
-    public InsertIntoTableCommand rewriteTvfParams(InsertIntoTableCommand originCommand, Offset nextOffset) {
-        // todo: only for cdc tvf
-        return null;
+    public InsertIntoTableCommand rewriteTvfParams(InsertIntoTableCommand originCommand, Offset nextOffset,
+            long taskId) {
+        throw new UnsupportedOperationException("rewriteTvfParams not supported for " + getClass().getSimpleName());
     }
 
     @Override
@@ -347,7 +360,7 @@ public Offset deserializeOffset(String offset) {
 
     @Override
     public Offset deserializeOffsetProperty(String offset) {
-        // no need
+        // no need cause cdc_stream has offset property
         return null;
     }
 
@@ -415,7 +428,7 @@ public void replayIfNeed(StreamingInsertJob job) throws JobException {
      * Assign the HW value to the synchronized Split,
      * and remove the Split from remainSplit and place it in finishedSplit.
      */
-    private List<SnapshotSplit> recalculateRemainingSplits(
+    protected List<SnapshotSplit> recalculateRemainingSplits(
             Map<String, Map<String, Map<String, String>>> chunkHighWatermarkMap,
             Map<String, List<SnapshotSplit>> snapshotSplits) {
         if (this.finishedSplits == null) {
@@ -541,7 +554,7 @@ private List<SnapshotSplit> requestTableSplits(String table) throws JobException
         }
     }
 
-    private boolean checkNeedSplitChunks(Map<String, String> sourceProperties) {
+    protected boolean checkNeedSplitChunks(Map<String, String> sourceProperties) {
         String startMode = sourceProperties.get(DataSourceConfigKeys.OFFSET);
         if (startMode == null) {
             return false;
@@ -550,11 +563,18 @@ private boolean checkNeedSplitChunks(Map<String, String> sourceProperties) {
                 || DataSourceConfigKeys.OFFSET_SNAPSHOT.equalsIgnoreCase(startMode);
     }
 
-    private boolean isSnapshotOnlyMode() {
+    protected boolean isSnapshotOnlyMode() {
         String offset = sourceProperties.get(DataSourceConfigKeys.OFFSET);
         return DataSourceConfigKeys.OFFSET_SNAPSHOT.equalsIgnoreCase(offset);
     }
 
+    @Override
+    public void onTaskCommitted(long scannedRows, long loadBytes) {
+        if (scannedRows == 0 && loadBytes == 0) {
+            hasMoreData = false;
+        }
+    }
+
     @Override
     public boolean hasReachedEnd() {
         return isSnapshotOnlyMode()