fix streaming job statistic never update

Author: sollhui

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;
@@ -417,8 +418,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(),

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()));
     }
 

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

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 {

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;
     }

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

@@ -377,7 +377,7 @@ private ExecutorFactory selectInsertExecutorFactory(
                             planner,
                             dataSink,
                             physicalSink,
-                            () -> new OlapInsertExecutor(ctx, olapTable, label, planner, insertCtx, emptyInsert)
+                            () -> new OlapInsertExecutor(ctx, olapTable, label, planner, insertCtx, emptyInsert, jobId)
                     );
                 }
 

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

@@ -89,6 +89,13 @@ public OlapInsertExecutor(ConnectContext ctx, Table table,
         this.olapTable = (OlapTable) table;
     }
 
+    public OlapInsertExecutor(ConnectContext ctx, Table table,
+            String labelName, NereidsPlanner planner, Optional<InsertCommandContext> insertCtx, boolean emptyInsert,
+            long jobId) {
+        super(ctx, table, labelName, planner, insertCtx, emptyInsert, jobId);
+        this.olapTable = (OlapTable) table;
+    }
+
     @Override
     public void beginTransaction() {
         if (isGroupCommitHttpStream()) {

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}