| | |
|---|
| | | { |
|---|
| | | requireTest(testId); |
|---|
| | | List<Map<String, Object>> source = new java.util.ArrayList<>(); |
|---|
| | | List<Map<String, Object>> snapshots = mapper.selectPersistedRealtimeSnapshots(testId); |
|---|
| | | if (snapshots != null && !snapshots.isEmpty()) source.addAll(snapshots); |
|---|
| | | else { |
|---|
| | | // 迁移完成前兼容旧数据；新采集只从统一快照读取。 |
|---|
| | | List<Map<String, Object>> cabinet = mapper.selectPersistedRecords(testId); |
|---|
| | | List<Map<String, Object>> collector = mapper.selectPersistedCollectorRecords(testId); |
|---|
| | | if (cabinet != null) source.addAll(cabinet); |
|---|
| | | if (collector != null) source.addAll(collector); |
|---|
| | | } |
|---|
| | | List<Map<String, Object>> effective = applyEffectiveData(testId, "REALTIME", source); |
|---|
| | | attachCollectorStepContext(effective); |
|---|
| | | return effective; |
|---|
| | |
|---|
| | | |
|---|
| | | private String stableIdentity(String dataType, Map<String, Object> row) |
|---|
| | | { |
|---|
| | | if ("REALTIME".equals(dataType) && "SNAPSHOT".equals(String.valueOf(row.get("sourceType")))) |
|---|
| | | return String.valueOf(row.getOrDefault("stableIdentity", "snapshot:" + value(row, "snapshotId"))); |
|---|
| | | if ("REALTIME".equals(dataType) && "COLLECTOR".equals(String.valueOf(row.get("sourceType")))) |
|---|
| | | return String.join(":", "collector", identityPart(row, "collectorId"), |
|---|
| | | identityPart(row, "channelCode"), value(row, "sequenceId"), identityPart(row, "metricCode")); |
|---|
