task-54: finalRowCount 改为任务内增量计数（upsert newlyInserted 累计），移除每 chunk 全表 COUNT(*)

2026-08-30 16:41:02 +08:00
parent e8cedbcb50
commit 91f760f789
3 changed files with 282 additions and 18 deletions
@@ -666,8 +666,12 @@ public class CollectDataService {
                requireRustfsPayload(storedDetail, "采集结果明细必须写入 RustFS");
                // 批量 upsert：先一次批量查现有行（payload_hash 相等即跳过，幂等），
                // 再按唯一键 uk_task_scope_item 分批发 INSERT ... ON DUPLICATE KEY UPDATE。
-                resultItemBatchWriter.upsertAccepted(task.getId(), result.getId(), scopeKey,
-                        chunkIndex, accepted, storedDetail);
+                // newlyInserted 是本 chunk 真实新增的行数，任务内增量累计得到 finalRowCount，
+                // 替代每个 chunk 一次全表 COUNT(*)，且与 chunk 乱序/重提无关。
+                CollectDataResultItemBatchWriter.UpsertCounts upsertCounts =
+                        resultItemBatchWriter.upsertAccepted(task.getId(), result.getId(), scopeKey,
+                                chunkIndex, accepted, storedDetail);
+                stats.finalRowCount += upsertCounts.newlyInserted();
            }

            String payloadJson = writeJson(rows, "采集结果序列化失败");
@@ -677,7 +681,6 @@ public class CollectDataService {
            persistChunk(taskId, scopeKey, scopeHash, chunkIndex, chunkTotal, storedPayload, payloadJson);
            persistScope(taskId, scopeKey, scopeHash, chunkTotal, request);

-            stats.finalRowCount = countFinalRows(taskId);
            // Python 在 done=true 那次回传携带关键词级聚合统计；非空时按"最后一次为准"覆盖。
            List<CollectDataSummaryRowDto> incomingSummaries = request.getSummaries();
            if (incomingSummaries != null && !incomingSummaries.isEmpty()) {
@@ -946,16 +949,6 @@ public class CollectDataService {
        return out;
    }

-    private int countFinalRows(Long taskId) {
-        if (taskId == null || taskId <= 0) {
-            return 0;
-        }
-        Long count = taskResultItemMapper.selectCount(new LambdaQueryWrapper<TaskResultItemEntity>()
-                .eq(TaskResultItemEntity::getTaskId, taskId)
-                .eq(TaskResultItemEntity::getModuleType, MODULE_TYPE));
-        return count == null ? 0 : count.intValue();
-    }
-
    private void ensureRustfsPayloadStorageEnabled() {
        if (!transientPayloadStorageService.isSharedWriteEnabled()) {
            throw new BusinessException("RustFS 未配置，采集结果回传暂不可接收");
@@ -1011,7 +1004,6 @@ public class CollectDataService {
                                                    CollectDataSubmitResultRequest request,
                                                    int currentChunkRows) {
        CollectDataStats stats = loadStats(task);
-        stats.finalRowCount = countFinalRows(task == null ? null : task.getId());
        CollectDataSubmitResultVo vo = new CollectDataSubmitResultVo();
        vo.setTaskId(task == null ? null : task.getId());
        vo.setResultId(result == null ? null : result.getId());
@@ -45,15 +45,21 @@ public class CollectDataResultItemBatchWriter {
        this.batchSize = batchSize <= 0 ? DEFAULT_BATCH_SIZE : batchSize;
    }

-    /** 批量写入计数：insertedOrUpdated 实际写入行数，skipped hash 相等跳过行数。 */
-    public record UpsertCounts(int insertedOrUpdated, int skipped) {
+    /**
+     * 批量写入计数：
+     * insertedOrUpdated 由 mapper 返回的 affected 行数累加（INSERT=1、存量更新可能为 2，
+     * 仅用于诊断日志）；
+     * newlyInserted 是本次调用真实新增的行数（仅原本不存在的行，hash 相等跳过与存量
+     * 更新均不计入，批量失败扣除未写入的新行），供调用方做任务内 finalRowCount 增量累计。
+     */
+    public record UpsertCounts(int insertedOrUpdated, int skipped, int newlyInserted) {
    }

    /** 把整 chunk 的 accepted 行批量 upsert；scopeKey 与 chunk 内 offset 已知，仅计算 refJson 与 hash。 */
    public UpsertCounts upsertAccepted(Long taskId, Long resultId, String scopeKey, int chunkIndex,
                                       List<CollectDataResultRowVo> rows, String storedDetail) {
        if (rows == null || rows.isEmpty()) {
-            return new UpsertCounts(0, 0);
+            return new UpsertCounts(0, 0, 0);
        }
        String scopeHash = sha256(scopeKey);
        // 一次性取回本 scope 现有行，构建 item_key → 现有行 映射（hash 相等即跳过）。
@@ -71,6 +77,9 @@ public class CollectDataResultItemBatchWriter {

        List<TaskResultItemEntity> toUpsert = new ArrayList<>();
        int skipped = 0;
+        // 仅原本不存在的行（真 INSERT）计入 newlyInserted；存量行 hash 不同触发 UPDATE
+        // 时表内行数不变，不计入，避免 finalRowCount 增量虚高。
+        int newlyInserted = 0;
        LocalDateTime now = LocalDateTime.now();
        // 批量生成整 chunk 的引用 JSON + hash（一次迭代），替代逐行 encodeRef + hash。
        List<CollectDataResultDetailCodec.RefWithHash> refsWithHash =
@@ -111,6 +120,9 @@ public class CollectDataResultItemBatchWriter {
            entity.setCreatedAt(existing == null ? now : existing.getCreatedAt());
            entity.setUpdatedAt(now);
            toUpsert.add(entity);
+            if (existing == null) {
+                newlyInserted++;
+            }
        }

        int written = 0;
@@ -122,9 +134,16 @@ public class CollectDataResultItemBatchWriter {
            } catch (RuntimeException ex) {
                log.warn("[collect-data] upsert result item batch failed, skip batch {}..{} taskId={}",
                        from, to, taskId, ex);
+                // 失败批的新行未落库，从增量计数中扣除，避免任务内累计虚高；
+                // 重提该 chunk 时按存量 hash 跳过已落库行、补插未落库行，累计收敛到真实行数。
+                for (TaskResultItemEntity entity : batch) {
+                    if (entity.getId() == null) {
+                        newlyInserted--;
+                    }
+                }
            }
        }
-        return new UpsertCounts(written, skipped);
+        return new UpsertCounts(written, skipped, newlyInserted);
    }

    private static String sha256(String value) {
@@ -0,0 +1,253 @@
+package com.nanri.aiimage.modules.collectdata.util;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.nanri.aiimage.modules.collectdata.model.vo.CollectDataResultRowVo;
+import com.nanri.aiimage.modules.task.mapper.TaskResultItemMapper;
+import com.nanri.aiimage.modules.task.model.entity.TaskResultItemEntity;
+import com.nanri.aiimage.modules.task.service.TransientPayloadStorageService;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyList;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * Task 54：将 finalRowCount 从每个 chunk COUNT(*) 改为任务内增量计数。
+ * upsertAccepted 返回的 newlyInserted 是本次调用真实新增的行数（仅
+ * existing == null 的真 INSERT，hash 相等跳过与存量更新均不计入），
+ * 调用方（CollectDataService）按任务内累加得到 finalRowCount，替代每
+ * chunk 一次全表 COUNT(*)；乱序提交下累计值与顺序无关，重提幂等不重复
+ * 计数，批量失败扣除未写入的新行。
+ */
+class CollectDataResultItemCountTest {
+
+    private TaskResultItemMapper taskResultItemMapper;
+    private TransientPayloadStorageService transientPayloadStorageService;
+    private CollectDataResultDetailCodec codec;
+    private CollectDataResultItemBatchWriter writer;
+
+    @BeforeEach
+    void setUp() {
+        taskResultItemMapper = mock(TaskResultItemMapper.class);
+        transientPayloadStorageService = mock(TransientPayloadStorageService.class);
+        codec = new CollectDataResultDetailCodec(new ObjectMapper());
+        writer = new CollectDataResultItemBatchWriter(taskResultItemMapper, codec, transientPayloadStorageService, 10);
+    }
+
+    @Test
+    void test_task_054_chunk_normal_default_path() {
+        // 正常路径：无现有行 → 2 行全部为真新增，newlyInserted=2（任务内累计基数）。
+        when(taskResultItemMapper.selectList(any())).thenReturn(List.of());
+        when(taskResultItemMapper.upsertBatch(anyList())).thenReturn(2);
+
+        CollectDataResultItemBatchWriter.UpsertCounts counts =
+                writer.upsertAccepted(1L, 2L, "task:1", 0,
+                        List.of(row("B000000001", "Nike"), row("B000000002", "Zara")), "rustfs:detail/abc");
+
+        assertEquals(2, counts.newlyInserted(), "无存量时全部为新行");
+        assertEquals(2, counts.insertedOrUpdated(), "写入计数一致");
+        assertEquals(0, counts.skipped(), "无跳过");
+    }
+
+    @Test
+    void test_task_054_chunk_normal_multiple_items() {
+        // 批量场景（任务内累计）：chunk0 已提交 B1/B2/B3 后，chunk1 再提交
+        // B3/B4/B5 → B3 跨 chunk 的 chunkIndex 不同 → 引用 hash 必然不同 → 走存量更新
+        // （行数不变，不计 newlyInserted）；B4/B5 真新增 → newlyInserted=2。
+        // 任务内累计 3+2=5 等于表内最终行数，不依赖 COUNT(*)。
+        List<TaskResultItemEntity> existing = new ArrayList<>();
+        existing.add(existingItem("asin:B000000001", 0, 0, "rustfs:detail/x"));
+        existing.add(existingItem("asin:B000000002", 1, 0, "rustfs:detail/x"));
+        existing.add(existingItem("asin:B000000003", 2, 0, "rustfs:detail/x"));
+        when(taskResultItemMapper.selectList(any())).thenReturn(existing);
+        when(taskResultItemMapper.upsertBatch(anyList())).thenAnswer(
+                invocation -> ((List<?>) invocation.getArgument(0)).size());
+
+        CollectDataResultItemBatchWriter.UpsertCounts counts =
+                writer.upsertAccepted(1L, 2L, "task:1", 1,
+                        List.of(row("B000000003", "Nike"), row("B000000004", "Zara"), row("B000000005", "Adidas")),
+                        "rustfs:detail/x");
+
+        assertEquals(2, counts.newlyInserted(), "仅 B4/B5 为真新增，B3 跨 chunk 更新不新增行");
+        assertEquals(3, counts.insertedOrUpdated(), "3 行均触发写入（B3 更新 + B4/B5 插入）");
+        assertEquals(0, counts.skipped(), "跨 chunk hash 不同无跳过");
+    }
+
+    @Test
+    void test_task_054_chunk_normal_repeated_operation_is_idempotent() {
+        // 幂等：同一 chunk 完全重提（现有行 hash 全部相等）→ newlyInserted=0，
+        // 不重复计数、不触发写入，finalRowCount 不虚高。
+        String refJson = codec.encodeRef(0, 0, "rustfs:detail/x");
+        TaskResultItemEntity existing = new TaskResultItemEntity();
+        existing.setId(100L);
+        existing.setItemKey("asin:B000000001");
+        existing.setPayloadJson(refJson);
+        existing.setPayloadHash(sha256(refJson));
+        when(taskResultItemMapper.selectList(any())).thenReturn(List.of(existing));
+
+        CollectDataResultItemBatchWriter.UpsertCounts counts =
+                writer.upsertAccepted(1L, 2L, "task:1", 0,
+                        List.of(row("B000000001", "Nike")), "rustfs:detail/x");
+
+        assertEquals(0, counts.newlyInserted(), "重提不新增行");
+        assertEquals(0, counts.insertedOrUpdated(), "无写入");
+        assertEquals(1, counts.skipped(), "全部跳过");
+        verify(taskResultItemMapper, never()).upsertBatch(anyList());
+    }
+
+    @Test
+    void test_task_054_chunk_boundary_empty_input() {
+        // 空输入：空列表零调用，newlyInserted=0，任务内累计不变。
+        CollectDataResultItemBatchWriter.UpsertCounts counts =
+                writer.upsertAccepted(1L, 2L, "task:1", 0, List.of(), "rustfs:x");
+
+        assertEquals(0, counts.newlyInserted(), "空输入不新增");
+        assertEquals(0, counts.insertedOrUpdated(), "无写入");
+        assertEquals(0, counts.skipped(), "无跳过");
+        verify(taskResultItemMapper, never()).selectList(any());
+    }
+
+    @Test
+    void test_task_054_chunk_boundary_single_item() {
+        // 单元素：单行真新增 → newlyInserted=1；单行存量更新（hash 不同）→ 0。
+        when(taskResultItemMapper.selectList(any())).thenReturn(List.of());
+        when(taskResultItemMapper.upsertBatch(anyList())).thenReturn(1);
+
+        CollectDataResultItemBatchWriter.UpsertCounts insert =
+                writer.upsertAccepted(1L, 2L, "task:1", 0,
+                        List.of(row("B000000001", "Nike")), "rustfs:new");
+
+        assertEquals(1, insert.newlyInserted(), "单行真新增计 1");
+
+        TaskResultItemEntity existing = new TaskResultItemEntity();
+        existing.setId(100L);
+        existing.setItemKey("asin:B000000001");
+        existing.setPayloadJson("{\"brand\":\"nike\",\"asin\":\"B000000001\"}");
+        existing.setPayloadHash("old-hash");
+        when(taskResultItemMapper.selectList(any())).thenReturn(List.of(existing));
+        when(taskResultItemMapper.upsertBatch(anyList())).thenReturn(1);
+
+        CollectDataResultItemBatchWriter.UpsertCounts update =
+                writer.upsertAccepted(1L, 2L, "task:1", 1,
+                        List.of(row("B000000001", "Nike")), "rustfs:new");
+
+        assertEquals(0, update.newlyInserted(), "存量行更新（hash 不同）不新增行");
+        assertEquals(1, update.insertedOrUpdated(), "写入发生但行数不变");
+    }
+
+    @Test
+    void test_task_054_chunk_boundary_limit_and_overflow() {
+        // 上限/超限：75 行 8 批次全部真新增 → newlyInserted=75，无无界积累。
+        when(taskResultItemMapper.selectList(any())).thenReturn(List.of());
+        when(taskResultItemMapper.upsertBatch(anyList())).thenAnswer(
+                invocation -> ((List<?>) invocation.getArgument(0)).size());
+        List<CollectDataResultRowVo> rows = new ArrayList<>();
+        for (int i = 0; i < 75; i++) {
+            rows.add(row("B" + String.format("%09d", i + 1), "brand-" + (i % 7)));
+        }
+
+        CollectDataResultItemBatchWriter.UpsertCounts counts =
+                writer.upsertAccepted(1L, 2L, "task:1", 3, rows, "rustfs:x");
+
+        assertEquals(75, counts.newlyInserted(), "全量真新增");
+        assertEquals(75, counts.insertedOrUpdated(), "写入计数一致");
+    }
+
+    @Test
+    void test_task_054_chunk_invalid_input_rejected() {
+        // 非法参数：null 行 / null ASIN / 空白 ASIN 均跳过且不计数 → newlyInserted 只含合法行。
+        when(taskResultItemMapper.selectList(any())).thenReturn(List.of());
+        when(taskResultItemMapper.upsertBatch(anyList())).thenAnswer(
+                invocation -> ((List<?>) invocation.getArgument(0)).size());
+        List<CollectDataResultRowVo> rows = new ArrayList<>();
+        rows.add(null);
+        rows.add(row(null, "nike"));
+        rows.add(row("   ", "zara"));
+        rows.add(row("B000000001", "nike"));
+
+        CollectDataResultItemBatchWriter.UpsertCounts counts =
+                writer.upsertAccepted(1L, 2L, "task:1", 0, rows, "rustfs:x");
+
+        assertEquals(1, counts.newlyInserted(), "仅合法行计数");
+        assertEquals(1, counts.insertedOrUpdated(), "仅合法行写入");
+    }
+
+    @Test
+    void test_task_054_chunk_dependency_failure_releases_resources() {
+        // 依赖失败：首批 upsertBatch 抛错 → 该批真新增行未落库，newlyInserted 扣除；
+        // 恢复后重提 → 已落库行跳过、未落库行补插，任务内累计仍等于表内真实行数。
+        when(taskResultItemMapper.selectList(any())).thenReturn(List.of());
+        doThrow(new RuntimeException("db down"))
+                .doAnswer(invocation -> ((List<?>) invocation.getArgument(0)).size())
+                .when(taskResultItemMapper).upsertBatch(anyList());
+        List<CollectDataResultRowVo> rows = new ArrayList<>();
+        for (int i = 0; i < 20; i++) {
+            rows.add(row("B" + String.format("%09d", i + 1), "brand"));
+        }
+
+        CollectDataResultItemBatchWriter.UpsertCounts first =
+                writer.upsertAccepted(1L, 2L, "task:1", 0, rows, "rustfs:x");
+
+        assertEquals(10, first.newlyInserted(), "首批失败扣除，仅第二批 10 行真新增");
+
+        // 重提同一 chunk：第二批 10 行 hash 相等跳过，首批 10 行补插 → newlyInserted=10。
+        List<TaskResultItemEntity> existing = new ArrayList<>();
+        for (int i = 10; i < 20; i++) {
+            String refJson = codec.encodeRef(0, i, "rustfs:x");
+            TaskResultItemEntity entity = new TaskResultItemEntity();
+            entity.setId(100L + i);
+            entity.setItemKey("asin:B" + String.format("%09d", i + 1));
+            entity.setPayloadJson(refJson);
+            entity.setPayloadHash(sha256(refJson));
+            existing.add(entity);
+        }
+        when(taskResultItemMapper.selectList(any())).thenReturn(existing);
+
+        CollectDataResultItemBatchWriter.UpsertCounts retry =
+                writer.upsertAccepted(1L, 2L, "task:1", 0, rows, "rustfs:x");
+
+        assertEquals(10, retry.newlyInserted(), "补插首批 10 行");
+        assertEquals(10, retry.skipped(), "第二批存量跳过");
+        assertEquals(20, first.newlyInserted() + retry.newlyInserted(), "任务内累计=表内真实行数");
+    }
+
+    private static TaskResultItemEntity existingItem(String itemKey, int offset, int chunkIndex, String pointer) {
+        TaskResultItemEntity entity = new TaskResultItemEntity();
+        entity.setId(100L + offset);
+        entity.setItemKey(itemKey);
+        String refJson = new CollectDataResultDetailCodec(new ObjectMapper()).encodeRef(chunkIndex, offset, pointer);
+        entity.setPayloadJson(refJson);
+        entity.setPayloadHash(sha256(refJson));
+        return entity;
+    }
+
+    private static CollectDataResultRowVo row(String asin, String brand) {
+        CollectDataResultRowVo row = new CollectDataResultRowVo();
+        row.setAsin(asin);
+        row.setBrand(brand);
+        return row;
+    }
+
+    private static String sha256(String value) {
+        try {
+            java.security.MessageDigest digest = java.security.MessageDigest.getInstance("SHA-256");
+            byte[] bytes = digest.digest((value == null ? "" : value).getBytes(java.nio.charset.StandardCharsets.UTF_8));
+            StringBuilder sb = new StringBuilder(bytes.length * 2);
+            for (byte b : bytes) {
+                sb.append(String.format("%02x", b));
+            }
+            return sb.toString();
+        } catch (Exception ex) {
+            throw new IllegalStateException(ex);
+        }
+    }
+}