From 3d085c16058ec9140a08204cc5e4b73c9e48329d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=BB=84=E8=87=AA=E8=BE=BE?= <980324341@qq.com> Date: Sat, 29 Aug 2026 16:17:12 +0800 Subject: [PATCH] =?UTF-8?q?task-11:=20Coze=20=E7=BB=93=E6=9E=9C=E5=90=88?= =?UTF-8?q?=E5=B9=B6=E5=89=8D=E6=8C=89=E7=A8=B3=E5=AE=9A=20rowKey=20HashSe?= =?UTF-8?q?t=20=E5=8E=BB=E9=87=8D=EF=BC=8C=E6=B6=88=E9=99=A4=20O(n=C2=B2)?= =?UTF-8?q?=20=E9=87=8D=E5=A4=8D=E5=A4=84=E7=90=86?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../service/SimilarAsinTaskService.java | 28 +- ...imilarAsinTaskServiceRowKeyDedupeTest.java | 324 ++++++++++++++++++ 2 files changed, 351 insertions(+), 1 deletion(-) create mode 100644 backend-java/src/test/java/com/nanri/aiimage/modules/similarasin/service/SimilarAsinTaskServiceRowKeyDedupeTest.java diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/service/SimilarAsinTaskService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/service/SimilarAsinTaskService.java index 6ae10510..42c32756 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/service/SimilarAsinTaskService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/service/SimilarAsinTaskService.java @@ -1533,6 +1533,30 @@ public class SimilarAsinTaskService { return index; } + /** + * Task 11：Coze 结果合并前按稳定 rowKey 一次性去重。 + * 优先 rowToken，缺失时回退归一化的 legacy key（id::ASIN::country）， + * 保留首次出现顺序。合并路径此前对每个重复行都重复 expand/分配/写回，是 O(n²) 热点。 + */ + List dedupeRowsByRowKey(List cozeRows) { + List deduped = new ArrayList<>(); + if (cozeRows == null || cozeRows.isEmpty()) { + return deduped; + } + Set seenKeys = new java.util.HashSet<>(); + for (SimilarAsinResultRowDto row : cozeRows) { + if (row == null) { + continue; + } + String key = rowKey(row); + if (key.isBlank() || !seenKeys.add(key)) { + continue; + } + deduped.add(row); + } + return deduped; + } + /** * Task 10：基于 rowKey 索引为 coze 回传行分配归属 chunk。 * 命中索引 → 归属该 chunk；未命中且有有效 fallback（chunkScopeHash + chunkIndex） @@ -3849,8 +3873,10 @@ public class SimilarAsinTaskService { rowsByChunk.put(chunkKey, readChunkRows(chunk)); chunkByKey.put(chunkKey, chunk); } + // Task 11：合并前按稳定 rowKey 去重，消除重复行逐行 expand/分配/写回 的 O(n²) 热点。 + List uniqueRows = dedupeRowsByRowKey(cozeRows); List expandedAll = new ArrayList<>(); - for (SimilarAsinResultRowDto resultRow : cozeRows) { + for (SimilarAsinResultRowDto resultRow : uniqueRows) { expandedAll.addAll(expandRows(List.of(resultRow), allRowsByBaseId)); } // Task 10：先建 rowKey → chunkKey 批量索引，把逐 chunk 线性扫描降为 O(1) 查找。 diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/similarasin/service/SimilarAsinTaskServiceRowKeyDedupeTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/similarasin/service/SimilarAsinTaskServiceRowKeyDedupeTest.java new file mode 100644 index 00000000..798c97b1 --- /dev/null +++ b/backend-java/src/test/java/com/nanri/aiimage/modules/similarasin/service/SimilarAsinTaskServiceRowKeyDedupeTest.java @@ -0,0 +1,324 @@ +package com.nanri.aiimage.modules.similarasin.service; + +import com.fasterxml.jackson.databind.ObjectMapper; +import com.nanri.aiimage.config.SimilarAsinProperties; +import com.nanri.aiimage.modules.similarasin.client.SimilarAsinCozeClient; +import com.nanri.aiimage.modules.similarasin.mapper.SimilarAsinFilterConditionMapper; +import com.nanri.aiimage.modules.similarasin.model.dto.SimilarAsinResultRowDto; +import com.nanri.aiimage.modules.similarasin.util.SimilarAsinImageEmbedder; +import com.nanri.aiimage.modules.file.service.LocalFileStorageService; +import com.nanri.aiimage.modules.task.mapper.FileResultMapper; +import com.nanri.aiimage.modules.task.mapper.FileTaskMapper; +import com.nanri.aiimage.modules.task.mapper.TaskChunkMapper; +import com.nanri.aiimage.modules.task.mapper.TaskScopeStateMapper; +import com.nanri.aiimage.modules.task.model.entity.FileTaskEntity; +import com.nanri.aiimage.modules.task.model.entity.TaskChunkEntity; +import com.nanri.aiimage.modules.task.service.TransientPayloadStorageService; +import com.baomidou.mybatisplus.core.MybatisConfiguration; +import com.baomidou.mybatisplus.core.metadata.TableInfoHelper; +import org.apache.ibatis.builder.MapperBuilderAssistant; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.InjectMocks; +import org.mockito.Mock; +import org.mockito.Spy; +import org.mockito.junit.jupiter.MockitoExtension; + +import java.lang.reflect.Method; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.concurrent.atomic.AtomicLong; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.lenient; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +/** + * Task 11：Coze 结果合并的重复检测从 O(n²) 改为 HashSet/稳定 row key。 + * dedupeRowsByRowKey 用 HashSet 按稳定 rowKey 一次性去重（保留顺序）， + * mergeCozeRowsIntoChunk 合并前先去重，消除重复行逐行重复处理。 + */ +@ExtendWith(MockitoExtension.class) +class SimilarAsinTaskServiceRowKeyDedupeTest { + + private static final AtomicLong NEXT_ID = new AtomicLong(70000); + + @Mock private LocalFileStorageService localFileStorageService; + @Mock private com.nanri.aiimage.modules.file.service.oss.OssStorageService ossStorageService; + @Mock private com.nanri.aiimage.config.StorageProperties storageProperties; + @Mock private FileTaskMapper fileTaskMapper; + @Mock private FileResultMapper fileResultMapper; + @Mock private TaskScopeStateMapper taskScopeStateMapper; + @Mock private TaskChunkMapper taskChunkMapper; + @Mock private SimilarAsinFilterConditionMapper filterConditionMapper; + @Spy private ObjectMapper objectMapper = new ObjectMapper(); + @Mock private SimilarAsinCozeClient cozeClient; + @Mock private SimilarAsinTaskCacheService taskCacheService; + @Mock private SimilarAsinProperties properties; + @Mock private com.nanri.aiimage.modules.task.service.TaskFileJobService taskFileJobService; + @Mock private com.nanri.aiimage.modules.task.service.TaskDistributedLockService taskDistributedLockService; + @Mock private com.nanri.aiimage.modules.task.service.TaskProgressSnapshotService taskProgressSnapshotService; + @Mock private TransientPayloadStorageService transientPayloadStorageService; + @Mock private org.springframework.transaction.PlatformTransactionManager transactionManager; + @Mock private com.nanri.aiimage.common.service.DistributedJobLockService distributedJobLockService; + @Mock private com.nanri.aiimage.config.InstanceMetadata instanceMetadata; + @Mock private com.nanri.aiimage.modules.coze.service.CozeCredentialPoolService cozeCredentialPoolService; + @Mock private SimilarAsinImageEmbedder imageEmbedder; + @Mock private SimilarAsinImagePrefetchService imagePrefetchService; + + @InjectMocks private SimilarAsinTaskService service; + + @BeforeAll + static void initializeMybatisMetadata() { + MapperBuilderAssistant assistant = new MapperBuilderAssistant(new MybatisConfiguration(), ""); + TableInfoHelper.initTableInfo(assistant, FileTaskEntity.class); + TableInfoHelper.initTableInfo(assistant, TaskChunkEntity.class); + } + + @BeforeEach + void setUp() { + lenient().when(properties.isBoundedResultAssemblyEnabled()).thenReturn(true); + lenient().when(properties.getImageCacheMaxBytes()).thenReturn(256L * 1024L * 1024L); + lenient().when(properties.getResultFileTimeoutMinutes()).thenReturn(90); + lenient().when(properties.getParseResponsePreviewLimit()).thenReturn(100); + lenient().when(properties.getMaxSourceFileBytes()).thenReturn(50L * 1024L * 1024L); + lenient().when(properties.getMaxParseRows()).thenReturn(50000); + lenient().when(properties.getMaxFieldLength()).thenReturn(2000); + lenient().when(properties.getMaxWorkbookZipEntries()).thenReturn(20000); + lenient().when(properties.getMaxWorkbookUncompressedBytes()).thenReturn(512L * 1024L * 1024L); + lenient().when(transientPayloadStorageService.storeParsedPayloadFast( + eq(SimilarAsinTaskService.MODULE_TYPE), any(), anyString(), anyString(), eq(false))) + .thenReturn("rustfs:task-parsed/similar-asin/70000/payload.json"); + lenient().doAnswer(invocation -> { + FileTaskEntity task = invocation.getArgument(0); + task.setId(NEXT_ID.incrementAndGet()); + return 1; + }).when(fileTaskMapper).insert(any(FileTaskEntity.class)); + lenient().when(fileResultMapper.insert(any(com.nanri.aiimage.modules.task.model.entity.FileResultEntity.class))).thenReturn(1); + lenient().when(taskScopeStateMapper.insert(any(com.nanri.aiimage.modules.task.model.entity.TaskScopeStateEntity.class))).thenReturn(1); + lenient().when(taskChunkMapper.selectList(any())).thenReturn(List.of()); + lenient().when(fileResultMapper.selectList(any())).thenReturn(List.of()); + } + + @AfterEach + void shutdown() { + service.shutdownAssembleExecutor(); + } + + private static SimilarAsinResultRowDto row(String rowToken, String id, String asin, String country) { + SimilarAsinResultRowDto r = new SimilarAsinResultRowDto(); + r.setRowToken(rowToken); + r.setId(id); + r.setAsin(asin); + r.setCountry(country); + return r; + } + + private static String rowsJson(List rows) throws Exception { + return new ObjectMapper().writeValueAsString(rows); + } + + private static TaskChunkEntity chunk(Long id, String scopeHash, Integer chunkIndex, String payloadJson) { + TaskChunkEntity chunk = new TaskChunkEntity(); + chunk.setId(id); + chunk.setTaskId(7004L); + chunk.setModuleType(SimilarAsinTaskService.MODULE_TYPE); + chunk.setScopeHash(scopeHash); + chunk.setChunkIndex(chunkIndex); + chunk.setPayloadJson(payloadJson); + return chunk; + } + + private void stubSingleChunkMerge(TaskChunkEntity chunk, String payloadJson, AtomicLong storedCounter) throws Exception { + when(taskChunkMapper.selectList(any())).thenReturn(List.of(chunk)); + when(transientPayloadStorageService.resolvePayload(eq(chunk.getPayloadJson()), anyString())) + .thenReturn(payloadJson); + when(transientPayloadStorageService.storeChunkPayloadVersioned(anyString(), any(), anyString(), any(), anyString())) + .thenAnswer(invocation -> { + storedCounter.incrementAndGet(); + return "stored:" + invocation.getArgument(2); + }); + when(taskChunkMapper.selectOne(any())).thenReturn(chunk); + when(taskChunkMapper.update(any(), any())).thenReturn(1); + } + + @Test + void test_task_011_merge_row_key_normal_default_path() throws Exception { + // 正常输入：cozeRows 含同一 rowKey 的重复行，merge 前按稳定 rowKey 去重， + // chunk payload 只写一次，结果行不重复。 + TaskChunkEntity chunk = chunk(1L, "hashA", 1, "ptr:chunk-A"); + AtomicLong storedCounter = new AtomicLong(0); + stubSingleChunkMerge(chunk, rowsJson(List.of(row("r1", "1", "B0A0000001", "英国"))), storedCounter); + + FileTaskEntity task = new FileTaskEntity(); + task.setId(7004L); + List cozeRows = List.of( + row("r1", "1", "B0A0000001", "英国"), + row("r1", "1", "B0A0000001", "英国")); + + List deduped = service.dedupeRowsByRowKey(cozeRows); + assertEquals(1, deduped.size(), "重复行必须按稳定 rowKey 去重"); + assertEquals("r1", deduped.get(0).getRowToken()); + + Method merge = SimilarAsinTaskService.class.getDeclaredMethod("mergeCozeRowsIntoChunk", + FileTaskEntity.class, String.class, Integer.class, List.class, Map.class); + merge.setAccessible(true); + merge.invoke(service, task, null, null, cozeRows, Map.of()); + assertEquals(1, storedCounter.get(), "去重后 chunk 只写一次"); + verify(transientPayloadStorageService, times(1)).storeChunkPayloadVersioned(anyString(), any(), anyString(), any(), anyString()); + } + + @Test + void test_task_011_merge_row_key_normal_multiple_items() throws Exception { + // 批量场景：多个重复行跨 chunk 分组，去重后顺序稳定、结果不丢失 + TaskChunkEntity chunkA = chunk(1L, "hashA", 1, "ptr:chunk-A"); + TaskChunkEntity chunkB = chunk(2L, "hashB", 2, "ptr:chunk-B"); + when(taskChunkMapper.selectList(any())).thenReturn(List.of(chunkA, chunkB)); + when(transientPayloadStorageService.resolvePayload(eq("ptr:chunk-A"), anyString())) + .thenReturn(rowsJson(List.of(row("r1", "1", "B0A0000001", "英国")))); + when(transientPayloadStorageService.resolvePayload(eq("ptr:chunk-B"), anyString())) + .thenReturn(rowsJson(List.of(row("r2", "2", "B0A0000002", "英国")))); + when(transientPayloadStorageService.storeChunkPayloadVersioned(anyString(), any(), anyString(), any(), anyString())) + .thenAnswer(invocation -> "stored:" + invocation.getArgument(2)); + AtomicLong selectOneRound = new AtomicLong(0); + when(taskChunkMapper.selectOne(any())).thenAnswer(invocation -> + selectOneRound.getAndIncrement() == 0 ? chunkA : chunkB); + when(taskChunkMapper.update(any(), any())).thenReturn(1); + + List cozeRows = new ArrayList<>(); + for (int i = 0; i < 3; i++) { + cozeRows.add(row("r1", "1", "B0A0000001", "英国")); + cozeRows.add(row("r2", "2", "B0A0000002", "英国")); + } + List deduped = service.dedupeRowsByRowKey(cozeRows); + assertEquals(2, deduped.size(), "3 轮重复输入去重后只剩 2 个唯一行"); + assertEquals(List.of("r1", "r2"), deduped.stream().map(SimilarAsinResultRowDto::getRowToken).toList(), + "去重必须保留首次出现顺序"); + + FileTaskEntity task = new FileTaskEntity(); + task.setId(7004L); + Method merge = SimilarAsinTaskService.class.getDeclaredMethod("mergeCozeRowsIntoChunk", + FileTaskEntity.class, String.class, Integer.class, List.class, Map.class); + merge.setAccessible(true); + merge.invoke(service, task, null, null, cozeRows, Map.of()); + verify(transientPayloadStorageService, times(2)).storeChunkPayloadVersioned(anyString(), any(), anyString(), any(), anyString()); + } + + @Test + void test_task_011_merge_row_key_normal_repeated_operation_is_idempotent() { + // 重复执行同一输入：去重结果完全一致，不产生重复记录 + List cozeRows = List.of( + row("r1", "1", "B0A0000001", "英国"), + row("r2", "2", "B0A0000002", "英国"), + row("r1", "1", "B0A0000001", "英国")); + List first = service.dedupeRowsByRowKey(cozeRows); + List second = service.dedupeRowsByRowKey(cozeRows); + assertEquals(first.size(), second.size()); + for (int i = 0; i < first.size(); i++) { + assertEquals(first.get(i).getRowToken(), second.get(i).getRowToken()); + assertEquals(first.get(i).getAsin(), second.get(i).getAsin()); + } + } + + @Test + void test_task_011_merge_row_key_boundary_empty_input() { + // 空输入：null/空列表安全返回空结果，不创建无效资源 + assertNotNull(service.dedupeRowsByRowKey(null)); + assertTrue(service.dedupeRowsByRowKey(null).isEmpty()); + assertTrue(service.dedupeRowsByRowKey(List.of()).isEmpty()); + // null 元素：跳过不抛异常 + List withNull = new ArrayList<>(); + withNull.add(null); + withNull.add(row("r1", "1", "B0A0000001", "英国")); + assertEquals(1, service.dedupeRowsByRowKey(withNull).size()); + } + + @Test + void test_task_011_merge_row_key_boundary_single_item() { + // 单行：不依赖批量路径，去重后结果正确 + List single = List.of(row("r1", "1", "B0A0000001", "英国")); + List deduped = service.dedupeRowsByRowKey(single); + assertEquals(1, deduped.size()); + assertEquals("r1", deduped.get(0).getRowToken()); + } + + @Test + void test_task_011_merge_row_key_boundary_limit_and_overflow() { + // 大批量：1000 行全部重复，去重后只剩 1 个唯一行，无无界内存增长 + List cozeRows = new ArrayList<>(); + for (int i = 0; i < 1000; i++) { + cozeRows.add(row("r1", "1", "B0A0000001", "英国")); + } + List deduped = service.dedupeRowsByRowKey(cozeRows); + assertEquals(1, deduped.size()); + // 1000 行唯一：全部保留且顺序稳定 + List unique = new ArrayList<>(); + for (int i = 0; i < 1000; i++) { + unique.add(row("r" + String.format("%04d", i), String.valueOf(i), "B0U" + String.format("%06d", i), "英国")); + } + List dedupedUnique = service.dedupeRowsByRowKey(unique); + assertEquals(1000, dedupedUnique.size()); + assertEquals("r0001", dedupedUnique.get(1).getRowToken()); + } + + @Test + void test_task_011_merge_row_key_invalid_input_rejected() { + // 稳定 rowKey 冲突：rowToken 为空时用 legacy key（id::ASIN::country）识别重复 + List noToken = List.of( + row("", "1", "B0A0000001", "英国"), + row("", "1", "b0a0000001", " 英国 ")); + List deduped = service.dedupeRowsByRowKey(noToken); + assertEquals(1, deduped.size(), "legacy key 归一化（ASIN 大写、country trim）后应识别为同一行"); + // 不同 ASIN：不误判为重复 + List diffAsin = List.of( + row("", "1", "B0A0000001", "英国"), + row("", "2", "B0A0000002", "英国")); + assertEquals(2, service.dedupeRowsByRowKey(diffAsin).size()); + } + + @Test + void test_task_011_merge_row_key_dependency_failure_releases_resources() throws Exception { + // chunk payload 读取失败：抛可识别业务异常且不写 chunk； + // 依赖恢复后重试成功，去重路径无残留状态 + TaskChunkEntity chunk = chunk(1L, "hashA", 1, "ptr:chunk-A"); + when(taskChunkMapper.selectList(any())).thenReturn(List.of(chunk)); + when(transientPayloadStorageService.resolvePayload(eq("ptr:chunk-A"), anyString())) + .thenThrow(new IllegalStateException("rustfs down")); + FileTaskEntity task = new FileTaskEntity(); + task.setId(7004L); + Method merge = SimilarAsinTaskService.class.getDeclaredMethod("mergeCozeRowsIntoChunk", + FileTaskEntity.class, String.class, Integer.class, List.class, Map.class); + merge.setAccessible(true); + Exception ex = assertThrows(Exception.class, () -> { + try { + merge.invoke(service, task, null, null, + List.of(row("r1", "1", "B0A0000001", "英国"), row("r1", "1", "B0A0000001", "英国")), Map.of()); + } catch (java.lang.reflect.InvocationTargetException e) { + throw e.getCause(); + } + }); + assertTrue(ex.getMessage() != null && ex.getMessage().contains("chunk"), + "chunk 读取失败消息必须可识别，实际: " + ex.getMessage()); + verify(transientPayloadStorageService, times(0)).storeChunkPayloadVersioned(anyString(), any(), anyString(), any(), anyString()); + // 恢复后重试成功：只写一次，无重复记录 + AtomicLong storedCounter = new AtomicLong(0); + stubSingleChunkMerge(chunk, rowsJson(List.of(row("r1", "1", "B0A0000001", "英国"))), storedCounter); + merge.invoke(service, task, null, null, + List.of(row("r1", "1", "B0A0000001", "英国"), row("r1", "1", "B0A0000001", "英国")), Map.of()); + assertEquals(1, storedCounter.get()); + verify(transientPayloadStorageService, times(1)).storeChunkPayloadVersioned(anyString(), any(), anyString(), any(), anyString()); + } +}