Revert "[fix](load) fix broker load progress due to retry"

This reverts commit c9ec9ec.
apache · Nov 5, 2024 · 26e1b06 · 26e1b06
1 parent a0ab4a3
commit 26e1b06
Show file tree

Hide file tree

Showing 2 changed files with 6 additions and 6 deletions.
diff --git a/fe/fe-core/src/main/java/org/apache/doris/load/loadv2/ProgressManager.java b/fe/fe-core/src/main/java/org/apache/doris/load/loadv2/ProgressManager.java
@@ -60,10 +60,10 @@ public void updateProgress(String id, TUniqueId queryId, TUniqueId fragmentId, i
         }
     }
 
-    public void setTotalScanNums(String id, int num) {
+    public void addTotalScanNums(String id, int num) {
         Progress progress = idToProgress.get(id);
         if (progress != null) {
-            progress.setTotalScanNums(num);
+            progress.addTotalScanNums(num);
         }
     }
 
@@ -87,8 +87,8 @@ static class Progress {
         private Table<TUniqueId, TUniqueId, Integer> finishedScanNums = HashBasedTable.create();
         private int totalScanNums = 0;
 
-        public synchronized void setTotalScanNums(int num) {
-            totalScanNums = num;
+        public synchronized void addTotalScanNums(int num) {
+            totalScanNums += num;
         }
 
         public synchronized void updateFinishedScanNums(TUniqueId queryId, TUniqueId fragmentId, int finishedScanNum) {

diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/Coordinator.java b/fe/fe-core/src/main/java/org/apache/doris/qe/Coordinator.java
@@ -628,7 +628,7 @@ public TPipelineFragmentParams getStreamLoadPlan() throws Exception {
         List<Long> relatedBackendIds = Lists.newArrayList(addressToBackendID.values());
         Env.getCurrentEnv().getLoadManager().initJobProgress(jobId, queryId, instanceIds,
                 relatedBackendIds);
-        Env.getCurrentEnv().getProgressManager().setTotalScanNums(String.valueOf(jobId), scanRangeNum);
+        Env.getCurrentEnv().getProgressManager().addTotalScanNums(String.valueOf(jobId), scanRangeNum);
         LOG.info("dispatch load job: {} to {}", DebugUtil.printId(queryId), addressToBackendID.keySet());
 
         Map<TNetworkAddress, TPipelineFragmentParams> tExecPlanFragmentParams
@@ -777,7 +777,7 @@ private void execInternal() throws Exception {
             List<Long> relatedBackendIds = Lists.newArrayList(addressToBackendID.values());
             Env.getCurrentEnv().getLoadManager().initJobProgress(jobId, queryId, instanceIds,
                     relatedBackendIds);
-            Env.getCurrentEnv().getProgressManager().setTotalScanNums(String.valueOf(jobId), scanRangeNum);
+            Env.getCurrentEnv().getProgressManager().addTotalScanNums(String.valueOf(jobId), scanRangeNum);
             LOG.info("dispatch load job: {} to {}", DebugUtil.printId(queryId), addressToBackendID.keySet());
         }