super/crawler-plugin
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

task-11: Coze 结果合并前按稳定 rowKey HashSet 去重，消除 O(n²) 重复处理

Browse Source
This commit is contained in:
黄自达
2026-08-29 16:17:12 +08:00
parent d64ba3a827
commit 3d085c1605
2 changed files with 351 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/service/SimilarAsinTaskService.java
+27 -1
View File
@@ -1533,6 +1533,30 @@ public class SimilarAsinTaskService {
return index;
}
/**
* Task 11：Coze 结果合并前按稳定 rowKey 一次性去重。
* 优先 rowToken，缺失时回退归一化的 legacy key（id::ASIN::country），
* 保留首次出现顺序。合并路径此前对每个重复行都重复 expand/分配/写回，是 O(n²) 热点。
*/
List<SimilarAsinResultRowDto> dedupeRowsByRowKey(List<SimilarAsinResultRowDto> cozeRows) {
List<SimilarAsinResultRowDto> deduped = new ArrayList<>();
if (cozeRows == null || cozeRows.isEmpty()) {
return deduped;
}
Set<String> 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<SimilarAsinResultRowDto> uniqueRows = dedupeRowsByRowKey(cozeRows);
List<SimilarAsinResultRowDto> 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) 查找。
backend-java/src/test/java/com/nanri/aiimage/modules/similarasin/service/SimilarAsinTaskServiceRowKeyDedupeTest.java
+324
View File
@@ -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<SimilarAsinResultRowDto> 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<SimilarAsinResultRowDto> cozeRows = List.of(
row("r1", "1", "B0A0000001", "英国"),
row("r1", "1", "B0A0000001", "英国"));
List<SimilarAsinResultRowDto> 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<SimilarAsinResultRowDto> cozeRows = new ArrayList<>();
for (int i = 0; i < 3; i++) {
cozeRows.add(row("r1", "1", "B0A0000001", "英国"));
cozeRows.add(row("r2", "2", "B0A0000002", "英国"));
}
List<SimilarAsinResultRowDto> 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<SimilarAsinResultRowDto> cozeRows = List.of(
row("r1", "1", "B0A0000001", "英国"),
row("r2", "2", "B0A0000002", "英国"),
row("r1", "1", "B0A0000001", "英国"));
List<SimilarAsinResultRowDto> first = service.dedupeRowsByRowKey(cozeRows);
List<SimilarAsinResultRowDto> 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<SimilarAsinResultRowDto> 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<SimilarAsinResultRowDto> single = List.of(row("r1", "1", "B0A0000001", "英国"));
List<SimilarAsinResultRowDto> 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<SimilarAsinResultRowDto> cozeRows = new ArrayList<>();
for (int i = 0; i < 1000; i++) {
cozeRows.add(row("r1", "1", "B0A0000001", "英国"));
}
List<SimilarAsinResultRowDto> deduped = service.dedupeRowsByRowKey(cozeRows);
assertEquals(1, deduped.size());
// 1000 行唯一：全部保留且顺序稳定
List<SimilarAsinResultRowDto> 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<SimilarAsinResultRowDto> 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<SimilarAsinResultRowDto> noToken = List.of(
row("", "1", "B0A0000001", "英国"),
row("", "1", "b0a0000001", " 英国 "));
List<SimilarAsinResultRowDto> deduped = service.dedupeRowsByRowKey(noToken);
assertEquals(1, deduped.size(), "legacy key 归一化（ASIN 大写、country trim）后应识别为同一行");
// 不同 ASIN：不误判为重复
List<SimilarAsinResultRowDto> 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());
}
}