diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/collectdata/service/CollectDataService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/collectdata/service/CollectDataService.java index 7fd1cef6..0cd1d586 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/collectdata/service/CollectDataService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/collectdata/service/CollectDataService.java @@ -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 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() - .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()); diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/collectdata/util/CollectDataResultItemBatchWriter.java b/backend-java/src/main/java/com/nanri/aiimage/modules/collectdata/util/CollectDataResultItemBatchWriter.java index 37f3895d..873ed1e8 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/collectdata/util/CollectDataResultItemBatchWriter.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/collectdata/util/CollectDataResultItemBatchWriter.java @@ -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 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 toUpsert = new ArrayList<>(); int skipped = 0; + // 仅原本不存在的行（真 INSERT）计入 newlyInserted；存量行 hash 不同触发 UPDATE + // 时表内行数不变，不计入，避免 finalRowCount 增量虚高。 + int newlyInserted = 0; LocalDateTime now = LocalDateTime.now(); // 批量生成整 chunk 的引用 JSON + hash（一次迭代），替代逐行 encodeRef + hash。 List 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) { diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/collectdata/util/CollectDataResultItemCountTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/collectdata/util/CollectDataResultItemCountTest.java new file mode 100644 index 00000000..aa1ffb1a --- /dev/null +++ b/backend-java/src/test/java/com/nanri/aiimage/modules/collectdata/util/CollectDataResultItemCountTest.java @@ -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 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 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 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 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 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); + } + } +}