[fix](statistics) fix updated rows incorrect due to typo in code

fe/fe-core/src/main/java/org/apache/doris/transaction/PublishVersionDaemon.java

@@ -129,7 +129,7 @@ private void publishVersion() {
             AgentTaskExecutor.submit(batchTask);
         }
 
-        Map<Long, Long> tableIdToNumDeltaRows = Maps.newHashMap();
+        Map<Long, Long> tableIdToTotalDeltaNumRows = Maps.newHashMap();
         // try to finish the transaction, if failed just retry in next loop
         for (TransactionState transactionState : readyTransactionStates) {
             Stream<PublishVersionTask> publishVersionTaskStream = transactionState
@@ -141,15 +141,15 @@ private void publishVersion() {
                             Map<Long, Long> tableIdToDeltaNumRows =
                                     task.getTableIdToDeltaNumRows();
                             tableIdToDeltaNumRows.forEach((tableId, numRows) -> {
-                                tableIdToDeltaNumRows
+                                tableIdToTotalDeltaNumRows
                                         .computeIfPresent(tableId, (id, orgNumRows) -> orgNumRows + numRows);
-                                tableIdToNumDeltaRows.putIfAbsent(tableId, numRows);
+                                tableIdToTotalDeltaNumRows.putIfAbsent(tableId, numRows);
                             });
                         }
                     });
             boolean hasBackendAliveAndUnfinishedTask = publishVersionTaskStream
                     .anyMatch(task -> !task.isFinished() && infoService.checkBackendAlive(task.getBackendId()));
-            transactionState.setTableIdToTotalNumDeltaRows(tableIdToNumDeltaRows);
+            transactionState.setTableIdToTotalNumDeltaRows(tableIdToTotalDeltaNumRows);
 
             boolean shouldFinishTxn = !hasBackendAliveAndUnfinishedTask || transactionState.isPublishTimeout()
                     || DebugPointUtil.isEnable("PublishVersionDaemon.not_wait_unfinished_tasks");

regression-test/suites/statistics/analyze_stats.groovy

@@ -1140,6 +1140,36 @@ PARTITION `p599` VALUES IN (599)
     afterDropped = sql """SHOW TABLE STATS test_meta_management"""
     assert check_column(afterDropped, "[col1, col2, col3]")
 
+    sql """ DROP TABLE IF EXISTS test_updated_rows """
+    sql """
+        CREATE TABLE test_updated_rows (
+            `col1` varchar(16) NOT NULL,
+            `col2` int(11) NOT NULL,
+            `col3` int(11) NOT NULL
+        ) ENGINE=OLAP
+        DUPLICATE KEY(`col1`)
+        DISTRIBUTED BY HASH(`col1`) BUCKETS 3
+        PROPERTIES (
+        "replication_allocation" = "tag.location.default: 1",
+        "storage_format" = "V2",
+        "light_schema_change" = "true",
+        "disable_auto_compaction" = "false",
+        "enable_single_replica_compaction" = "false"
+        );
+    """
+
+    sql """ANALYZE TABLE test_updated_rows WITH SYNC"""
+    sql """ INSERT INTO test_updated_rows VALUES('1',1,1); """
+    sql """ANALYZE TABLE test_updated_rows WITH SYNC"""
+    def cnt1 = sql """ SHOW TABLE STATS test_updated_rows """
+    assertEquals(Integer.valueOf(cnt1[0][0]), 1)
+    sql """ INSERT INTO test_updated_rows SELECT * FROM test_updated_rows """
+    sql """ INSERT INTO test_updated_rows SELECT * FROM test_updated_rows """
+    sql """ INSERT INTO test_updated_rows SELECT * FROM test_updated_rows """
+    sql """ANALYZE TABLE test_updated_rows WITH SYNC"""
+    def cnt2 = sql """ SHOW TABLE STATS test_updated_rows """
+    assertEquals(Integer.valueOf(cnt2[0][0]), 8)
+
     // test analyze specific column
     sql """CREATE TABLE test_analyze_specific_column (col1 varchar(11451) not null, col2 int not null, col3 int not null)
     DUPLICATE KEY(col1)
@@ -1213,6 +1243,4 @@ PARTITION `p599` VALUES IN (599)
     }
 
     assert all_finished(show_result)
-
-
 }