fix streaming job statistic never update

sollhui · sollhui · commit b5ed4a348128 · 2025-10-09T18:00:26.000+08:00
diff --git a/fe/fe-core/src/main/java/org/apache/doris/cloud/transaction/CloudGlobalTransactionMgr.java b/fe/fe-core/src/main/java/org/apache/doris/cloud/transaction/CloudGlobalTransactionMgr.java
@@ -623,16 +623,18 @@ private void commitTransactionWithoutLock(long dbId, List<Table> tableList, long
             } else if (txnCommitAttachment instanceof StreamingTaskTxnCommitAttachment) {
                 StreamingTaskTxnCommitAttachment streamingTaskTxnCommitAttachment =
                             (StreamingTaskTxnCommitAttachment) txnCommitAttachment;
-                TxnStateChangeCallback cb = callbackFactory.getCallback(streamingTaskTxnCommitAttachment.getTaskId());
+                TxnStateChangeCallback cb = callbackFactory.getCallback(streamingTaskTxnCommitAttachment.getJobId());
+                TxnCommitAttachment commitAttachment = null;
                 if (cb != null) {
                     // use a temporary transaction state to do before commit check,
                     // what actually works is the transactionId
                     TransactionState tmpTxnState = new TransactionState();
                     tmpTxnState.setTransactionId(transactionId);
                     cb.beforeCommitted(tmpTxnState);
+                    commitAttachment = tmpTxnState.getTxnCommitAttachment();
                 }
                 builder.setCommitAttachment(TxnUtil
-                        .streamingTaskTxnCommitAttachmentToPb(streamingTaskTxnCommitAttachment));
+                        .streamingTaskTxnCommitAttachmentToPb((StreamingTaskTxnCommitAttachment) commitAttachment));
             } else {
                 throw new UserException("invalid txnCommitAttachment");
             }
@@ -676,6 +678,11 @@ private void executeCommitTxnRequest(CommitTxnRequest commitTxnRequest, long tra
             if (txnCommitAttachment != null && txnCommitAttachment instanceof RLTaskTxnCommitAttachment) {
                 RLTaskTxnCommitAttachment rlTaskTxnCommitAttachment = (RLTaskTxnCommitAttachment) txnCommitAttachment;
                 callbackId = rlTaskTxnCommitAttachment.getJobId();
+            } else if (txnCommitAttachment != null
+                        && txnCommitAttachment instanceof StreamingTaskTxnCommitAttachment) {
+                StreamingTaskTxnCommitAttachment streamingTaskTxnCommitAttachment =
+                        (StreamingTaskTxnCommitAttachment) txnCommitAttachment;
+                callbackId = streamingTaskTxnCommitAttachment.getJobId();
             } else if (txnState != null) {
                 callbackId = txnState.getCallbackId();
             }
diff --git a/fe/fe-core/src/main/java/org/apache/doris/cloud/transaction/TxnUtil.java b/fe/fe-core/src/main/java/org/apache/doris/cloud/transaction/TxnUtil.java
@@ -373,6 +373,12 @@ public static TransactionState transactionStateFromPb(TxnInfoPB txnInfo) {
                 commitAttachment =
                     TxnUtil.rtTaskTxnCommitAttachmentFromPb(txnInfo.getCommitAttachment());
             }
+
+            if (txnInfo.getCommitAttachment().getType()
+                    == TxnCommitAttachmentPB.Type.STREAMING_TASK_TXN_COMMIT_ATTACHMENT) {
+                commitAttachment =
+                    TxnUtil.streamingTaskTxnCommitAttachmentFromPb(txnInfo.getCommitAttachment());
+            }
         }
 
         long prepareTime = txnInfo.hasPrepareTime() ? txnInfo.getPrepareTime() : -1;
diff --git a/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJob.java b/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJob.java
@@ -41,6 +41,7 @@
 import org.apache.doris.job.extensions.insert.InsertTask;
 import org.apache.doris.job.offset.SourceOffsetProvider;
 import org.apache.doris.job.offset.SourceOffsetProviderFactory;
+import org.apache.doris.load.loadv2.LoadJob;
 import org.apache.doris.load.loadv2.LoadStatistic;
 import org.apache.doris.nereids.StatementContext;
 import org.apache.doris.nereids.analyzer.UnboundTVFRelation;
@@ -295,8 +296,6 @@ public void onStreamTaskSuccess(StreamingInsertTask task) {
             Env.getCurrentEnv().getJobManager().getStreamingTaskManager().removeRunningTask(task);
             StreamingInsertTask nextTask = createStreamingInsertTask();
             this.runningStreamTask = nextTask;
-            //todo: maybe fetch from txn attachment?
-            offsetProvider.updateOffset(task.getRunningOffset());
         } finally {
             writeUnlock();
         }
@@ -417,8 +416,12 @@ public void beforeCommitted(TransactionState txnState) throws TransactionExcepti
             taskIds.add(runningStreamTask.getTaskId());
             // todo: Check whether the taskid of runningtask is consistent with the taskid associated with txn
 
-            // todo: need get loadStatistic, load manager statistic is empty
-            LoadStatistic loadStatistic = new LoadStatistic();
+            List<LoadJob> loadJobs = Env.getCurrentEnv().getLoadManager().queryLoadJobsByJobIds(taskIds);
+            if (loadJobs.size() == 0) {
+                throw new TransactionException("load job not found, insert job id is " + runningStreamTask.getTaskId());
+            }
+            LoadJob loadJob = loadJobs.get(0);
+            LoadStatistic loadStatistic = loadJob.getLoadStatistic();
             txnState.setTxnCommitAttachment(new StreamingTaskTxnCommitAttachment(
                         getJobId(),
                         runningStreamTask.getTaskId(),
diff --git a/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertTask.java b/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertTask.java
@@ -127,6 +127,7 @@ private void before() throws Exception {
 
         this.runningOffset = offsetProvider.getNextOffset(jobProperties, originTvfProps);
         InsertIntoTableCommand baseCommand = (InsertIntoTableCommand) new NereidsParser().parseSingle(sql);
+        baseCommand.setJobId(getTaskId());
         StmtExecutor baseStmtExecutor =
                 new StmtExecutor(ctx, new LogicalPlanAdapter(baseCommand, ctx.getStatementContext()));
         baseCommand.initPlan(ctx, baseStmtExecutor, false);
@@ -135,7 +136,6 @@ private void before() throws Exception {
         }
         this.taskCommand = offsetProvider.rewriteTvfParams(baseCommand, runningOffset);
         this.taskCommand.setLabelName(Optional.of(getJobId() + LABEL_SPLITTER + getTaskId()));
-        this.taskCommand.setJobId(getTaskId());
         this.stmtExecutor = new StmtExecutor(ctx, new LogicalPlanAdapter(taskCommand, ctx.getStatementContext()));
     }
 
diff --git a/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingTaskTxnCommitAttachment.java b/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingTaskTxnCommitAttachment.java
@@ -23,9 +23,14 @@
 
 import com.google.gson.annotations.SerializedName;
 import lombok.Getter;
+import lombok.Setter;
 
 public class StreamingTaskTxnCommitAttachment extends TxnCommitAttachment {
 
+    public StreamingTaskTxnCommitAttachment() {
+        super(TransactionState.LoadJobSourceType.STREAMING_JOB);
+    }
+
     public StreamingTaskTxnCommitAttachment(long jobId, long taskId,
                 long scannedRows, long loadBytes, long numFiles, long fileBytes, String offset) {
         super(TransactionState.LoadJobSourceType.STREAMING_JOB);
@@ -48,6 +53,7 @@ public StreamingTaskTxnCommitAttachment(StreamingTaskCommitAttachmentPB pb) {
     }
 
     @Getter
+    @Setter
     private long jobId;
     @Getter
     private long taskId;
diff --git a/fe/fe-core/src/main/java/org/apache/doris/job/offset/s3/S3SourceOffsetProvider.java b/fe/fe-core/src/main/java/org/apache/doris/job/offset/s3/S3SourceOffsetProvider.java
@@ -139,8 +139,10 @@ public InsertIntoTableCommand rewriteTvfParams(InsertIntoTableCommand originComm
             }
             return plan;
         });
-        return new InsertIntoTableCommand((LogicalPlan) rewritePlan, Optional.empty(), Optional.empty(),
-                Optional.empty(), true, Optional.empty());
+        InsertIntoTableCommand insertIntoTableCommand = new InsertIntoTableCommand((LogicalPlan) rewritePlan,
+                Optional.empty(), Optional.empty(), Optional.empty(), true, Optional.empty());
+        insertIntoTableCommand.setJobId(originCommand.getJobId());
+        return insertIntoTableCommand;
     }
 
     @Override
diff --git a/fe/fe-core/src/main/java/org/apache/doris/load/loadv2/InsertLoadJob.java b/fe/fe-core/src/main/java/org/apache/doris/load/loadv2/InsertLoadJob.java
@@ -57,6 +57,10 @@ public InsertLoadJob(long dbId, String label) {
         super(EtlJobType.INSERT, dbId, label);
     }
 
+    public InsertLoadJob(long dbId, String label, long jobId) {
+        super(EtlJobType.INSERT, dbId, label, jobId);
+    }
+
     public InsertLoadJob(String label, long transactionId, long dbId, long tableId,
             long createTimestamp, String failMsg, String trackingUrl, String firstErrorMsg,
             UserIdentity userInfo) throws MetaNotFoundException {
diff --git a/fe/fe-core/src/main/java/org/apache/doris/load/loadv2/LoadManager.java b/fe/fe-core/src/main/java/org/apache/doris/load/loadv2/LoadManager.java
@@ -194,6 +194,7 @@ protected void createLoadJob(LoadJob loadJob) {
 
     public void addLoadJob(LoadJob loadJob) {
         // Insert label may be null in txn mode, we add txn insert job after success.
+        LOG.info("add load job: {}, label: {}", loadJob.getId(), loadJob.getLabel());
         if (loadJob.getLabel() != null) {
             idToLoadJob.put(loadJob.getId(), loadJob);
             long dbId = loadJob.getDbId();
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/AbstractInsertExecutor.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/AbstractInsertExecutor.java
@@ -71,7 +71,7 @@ public abstract class AbstractInsertExecutor {
     protected long txnId = INVALID_TXN_ID;
 
     /**
-     * Constructor
+     * Randomly generate job ID.
      */
     public AbstractInsertExecutor(ConnectContext ctx, TableIf table, String labelName, NereidsPlanner planner,
             Optional<InsertCommandContext> insertCtx, boolean emptyInsert) {
@@ -87,6 +87,24 @@ public AbstractInsertExecutor(ConnectContext ctx, TableIf table, String labelNam
         this.emptyInsert = emptyInsert;
     }
 
+    /**
+     * Specify job ID.
+     */
+    public AbstractInsertExecutor(ConnectContext ctx, TableIf table, String labelName, NereidsPlanner planner,
+            Optional<InsertCommandContext> insertCtx, boolean emptyInsert, long jobId) {
+        this.ctx = ctx;
+        this.database = table.getDatabase();
+        this.insertLoadJob = new InsertLoadJob(database.getId(), labelName, jobId);
+        ctx.getEnv().getLoadManager().addLoadJob(insertLoadJob);
+        this.coordinator = EnvFactory.getInstance().createCoordinator(
+                ctx, planner, ctx.getStatsErrorEstimator(), insertLoadJob.getId());
+        this.labelName = labelName;
+        this.table = table;
+        this.insertCtx = insertCtx;
+        this.emptyInsert = emptyInsert;
+        this.jobId = jobId;
+    }
+
     public Coordinator getCoordinator() {
         return coordinator;
     }
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertIntoTableCommand.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertIntoTableCommand.java
@@ -132,6 +132,7 @@ public InsertIntoTableCommand(LogicalPlan logicalQuery, Optional<String> labelNa
         this.cte = cte;
         this.needNormalizePlan = needNormalizePlan;
         this.branchName = branchName;
+        this.jobId = Env.getCurrentEnv().getNextId();
     }
 
     /**
@@ -356,7 +357,8 @@ private ExecutorFactory selectInsertExecutorFactory(
                             planner,
                             dataSink,
                             physicalSink,
-                            () -> new OlapTxnInsertExecutor(ctx, olapTable, label, planner, insertCtx, emptyInsert)
+                            () -> new OlapTxnInsertExecutor(
+                                    ctx, olapTable, label, planner, insertCtx, emptyInsert, jobId)
                     );
                 } else if (ctx.isGroupCommit()) {
                     Backend groupCommitBackend = Env.getCurrentEnv()
@@ -369,15 +371,15 @@ private ExecutorFactory selectInsertExecutorFactory(
                             dataSink,
                             physicalSink,
                             () -> new OlapGroupCommitInsertExecutor(
-                                    ctx, olapTable, label, planner, insertCtx, emptyInsert, groupCommitBackend
+                                    ctx, olapTable, label, planner, insertCtx, emptyInsert, groupCommitBackend, jobId
                             )
                     );
                 } else {
                     executorFactory = ExecutorFactory.from(
                             planner,
                             dataSink,
                             physicalSink,
-                            () -> new OlapInsertExecutor(ctx, olapTable, label, planner, insertCtx, emptyInsert)
+                            () -> new OlapInsertExecutor(ctx, olapTable, label, planner, insertCtx, emptyInsert, jobId)
                     );
                 }
 
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/OlapGroupCommitInsertExecutor.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/OlapGroupCommitInsertExecutor.java
@@ -65,8 +65,8 @@ public class OlapGroupCommitInsertExecutor extends OlapInsertExecutor {
 
     public OlapGroupCommitInsertExecutor(ConnectContext ctx, Table table,
             String labelName, NereidsPlanner planner, Optional<InsertCommandContext> insertCtx,
-            boolean emptyInsert, Backend backend) {
-        super(ctx, table, labelName, planner, insertCtx, emptyInsert);
+            boolean emptyInsert, Backend backend, long jobId) {
+        super(ctx, table, labelName, planner, insertCtx, emptyInsert, jobId);
         this.groupCommitBackend = backend;
     }
 
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/OlapInsertExecutor.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/OlapInsertExecutor.java
@@ -32,6 +32,7 @@
 import org.apache.doris.common.util.DebugPointUtil;
 import org.apache.doris.common.util.DebugUtil;
 import org.apache.doris.job.extensions.insert.streaming.StreamingInsertTask;
+import org.apache.doris.job.extensions.insert.streaming.StreamingTaskTxnCommitAttachment;
 import org.apache.doris.load.EtlJobType;
 import org.apache.doris.nereids.NereidsPlanner;
 import org.apache.doris.nereids.exceptions.AnalysisException;
@@ -59,6 +60,7 @@
 import org.apache.doris.transaction.TransactionState.TxnCoordinator;
 import org.apache.doris.transaction.TransactionState.TxnSourceType;
 import org.apache.doris.transaction.TransactionStatus;
+import org.apache.doris.transaction.TxnCommitAttachment;
 
 import com.google.common.base.Strings;
 import com.google.common.collect.ImmutableList;
@@ -84,8 +86,9 @@ public class OlapInsertExecutor extends AbstractInsertExecutor {
      * constructor
      */
     public OlapInsertExecutor(ConnectContext ctx, Table table,
-            String labelName, NereidsPlanner planner, Optional<InsertCommandContext> insertCtx, boolean emptyInsert) {
-        super(ctx, table, labelName, planner, insertCtx, emptyInsert);
+            String labelName, NereidsPlanner planner, Optional<InsertCommandContext> insertCtx, boolean emptyInsert,
+            long jobId) {
+        super(ctx, table, labelName, planner, insertCtx, emptyInsert, jobId);
         this.olapTable = (OlapTable) table;
     }
 
@@ -190,6 +193,7 @@ protected void beforeExec() {
 
     @Override
     protected void onComplete() throws UserException {
+        TxnCommitAttachment txnCommitAttachment = buildTxnAttachment();
         setTxnCallbackId();
         if (ctx.getState().getStateType() == MysqlStateType.ERR) {
             try {
@@ -204,7 +208,7 @@ protected void onComplete() throws UserException {
                 database, Lists.newArrayList((Table) table),
                 txnId,
                 TabletCommitInfo.fromThrift(coordinator.getCommitInfos()),
-                ctx.getSessionVariable().getInsertVisibleTimeoutMs())) {
+                ctx.getSessionVariable().getInsertVisibleTimeoutMs(), txnCommitAttachment)) {
             txnStatus = TransactionStatus.VISIBLE;
         } else {
             txnStatus = TransactionStatus.COMMITTED;
@@ -231,13 +235,31 @@ private void setTxnCallbackId() {
         if (state == null) {
             throw new AnalysisException("txn does not exist: " + txnId);
         }
-        StreamingInsertTask streamingInsertTask =
-                Env.getCurrentEnv().getJobManager().getStreamingTaskManager().getStreamingInsertTaskById(jobId);
-        if (streamingInsertTask != null) {
-            state.setCallbackId(streamingInsertTask.getJobId());
+        StreamingInsertTask task = Env.getCurrentEnv()
+                                      .getJobManager()
+                                      .getStreamingTaskManager()
+                                      .getStreamingInsertTaskById(jobId);
+        if (task != null) {
+            state.setCallbackId(task.getJobId());
         }
     }
 
+    private TxnCommitAttachment buildTxnAttachment() {
+        if (!Config.isCloudMode()) {
+            return null;
+        }
+        StreamingInsertTask task = Env.getCurrentEnv()
+                                      .getJobManager()
+                                      .getStreamingTaskManager()
+                                      .getStreamingInsertTaskById(jobId);
+        if (task == null) {
+            return null;
+        }
+        StreamingTaskTxnCommitAttachment attachment = new StreamingTaskTxnCommitAttachment();
+        attachment.setJobId(task.getJobId());
+        return attachment;
+    }
+
     @Override
     protected void onFail(Throwable t) {
         errMsg = t.getMessage() == null ? "unknown reason" : t.getMessage();
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/OlapTxnInsertExecutor.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/OlapTxnInsertExecutor.java
@@ -42,8 +42,8 @@ public class OlapTxnInsertExecutor extends OlapInsertExecutor {
 
     public OlapTxnInsertExecutor(ConnectContext ctx, Table table,
             String labelName, NereidsPlanner planner, Optional<InsertCommandContext> insertCtx,
-            boolean emptyInsert) {
-        super(ctx, table, labelName, planner, insertCtx, emptyInsert);
+            boolean emptyInsert, long jobId) {
+        super(ctx, table, labelName, planner, insertCtx, emptyInsert, jobId);
         txnStatus = TransactionStatus.PREPARE;
     }
 
diff --git a/regression-test/suites/job_p0/streaming_job/test_streaming_insert_job.groovy b/regression-test/suites/job_p0/streaming_job/test_streaming_insert_job.groovy
@@ -65,22 +65,22 @@ suite("test_streaming_insert_job") {
         Awaitility.await().atMost(300, SECONDS)
                 .pollInterval(1, SECONDS).until(
                 {
-                    print("check success task count")
                     def jobSuccendCount = sql """ select SucceedTaskCount from jobs("type"="insert") where Name like '%${jobName}%' and ExecuteType='STREAMING' """
+                    log.info("jobSuccendCount: " + jobSuccendCount)
                     // check job status and succeed task count larger than 2
                     jobSuccendCount.size() == 1 && '2' <= jobSuccendCount.get(0).get(0)
                 }
         )
     } catch (Exception ex){
         def showjob = sql """select * from jobs("type"="insert") where Name='${jobName}'"""
         def showtask = sql """select * from tasks("type"="insert") where JobName='${jobName}'"""
-        println("show job: " + showjob)
-        println("show task: " + showtask)
+        log.info("show job: " + showjob)
+        log.info("show task: " + showtask)
         throw ex;
     }
 
     def jobResult = sql """select * from jobs("type"="insert") where Name='${jobName}'"""
-    println("show success job: " + jobResult)
+    log.info("show success job: " + jobResult)
 
     qt_select """ SELECT * FROM ${tableName} order by c1 """
 
@@ -96,12 +96,14 @@ suite("test_streaming_insert_job") {
     assert pauseShowTask.size() == 0
 
 
-    def jobOffset = sql """
-        select currentOffset, endoffset from jobs("type"="insert") where Name='${jobName}'
+    def jobInfo = sql """
+        select currentOffset, endoffset, loadStatistic from jobs("type"="insert") where Name='${jobName}'
     """
-    assert jobOffset.get(0).get(0) == "{\"endFile\":\"regression/load/data/example_1.csv\"}";
-    assert jobOffset.get(0).get(1) == "{\"endFile\":\"regression/load/data/example_1.csv\"}";
-
+    log.info("jobInfo: " + jobInfo)
+    assert jobInfo.get(0).get(0) == "{\"endFile\":\"regression/load/data/example_1.csv\"}";
+    assert jobInfo.get(0).get(1) == "{\"endFile\":\"regression/load/data/example_1.csv\"}";
+    assert jobInfo.get(0).get(2) == "{\"scannedRows\":20,\"loadBytes\":425,\"fileNumber\":0,\"fileSize\":0}"
+    
     // alter streaming job
     sql """
        ALTER JOB ${jobName}

Original file line number	Diff line number	Diff line change
`@@ -373,6 +373,12 @@ public static TransactionState transactionStateFromPb(TxnInfoPB txnInfo) {`
`373`	`373`	`commitAttachment =`
`374`	`374`	`TxnUtil.rtTaskTxnCommitAttachmentFromPb(txnInfo.getCommitAttachment());`
`375`	`375`	`}`
	`376`	`+`
	`377`	`+ if (txnInfo.getCommitAttachment().getType()`
	`378`	`+ == TxnCommitAttachmentPB.Type.STREAMING_TASK_TXN_COMMIT_ATTACHMENT) {`
	`379`	`+ commitAttachment =`
	`380`	`+ TxnUtil.streamingTaskTxnCommitAttachmentFromPb(txnInfo.getCommitAttachment());`
	`381`	`+ }`
`376`	`382`	`}`
`377`	`383`
`378`	`384`	`long prepareTime = txnInfo.hasPrepareTime() ? txnInfo.getPrepareTime() : -1;`
Original file line number	Diff line number	Diff line change
`@@ -127,6 +127,7 @@ private void before() throws Exception {`
`127`	`127`
`128`	`128`	`this.runningOffset = offsetProvider.getNextOffset(jobProperties, originTvfProps);`
`129`	`129`	`InsertIntoTableCommand baseCommand = (InsertIntoTableCommand) new NereidsParser().parseSingle(sql);`
	`130`	`+ baseCommand.setJobId(getTaskId());`
`130`	`131`	`StmtExecutor baseStmtExecutor =`
`131`	`132`	`new StmtExecutor(ctx, new LogicalPlanAdapter(baseCommand, ctx.getStatementContext()));`
`132`	`133`	`baseCommand.initPlan(ctx, baseStmtExecutor, false);`
`@@ -135,7 +136,6 @@ private void before() throws Exception {`
`135`	`136`	`}`
`136`	`137`	`this.taskCommand = offsetProvider.rewriteTvfParams(baseCommand, runningOffset);`
`137`	`138`	`this.taskCommand.setLabelName(Optional.of(getJobId() + LABEL_SPLITTER + getTaskId()));`
`138`		`- this.taskCommand.setJobId(getTaskId());`
`139`	`139`	`this.stmtExecutor = new StmtExecutor(ctx, new LogicalPlanAdapter(taskCommand, ctx.getStatementContext()));`
`140`	`140`	`}`
`141`	`141`
Original file line number	Diff line number	Diff line change
`@@ -139,8 +139,10 @@ public InsertIntoTableCommand rewriteTvfParams(InsertIntoTableCommand originComm`
`139`	`139`	`}`
`140`	`140`	`return plan;`
`141`	`141`	`});`
`142`		`- return new InsertIntoTableCommand((LogicalPlan) rewritePlan, Optional.empty(), Optional.empty(),`
`143`		`- Optional.empty(), true, Optional.empty());`
	`142`	`+ InsertIntoTableCommand insertIntoTableCommand = new InsertIntoTableCommand((LogicalPlan) rewritePlan,`
	`143`	`+ Optional.empty(), Optional.empty(), Optional.empty(), true, Optional.empty());`
	`144`	`+ insertIntoTableCommand.setJobId(originCommand.getJobId());`
	`145`	`+ return insertIntoTableCommand;`
`144`	`146`	`}`
`145`	`147`
`146`	`148`	`@Override`