-
feat(sync): PR4 — bulk createMany + batched parser(9h→30min 性能优化) · 3fd28974
Hot path 优化 — 把每行一次 SQL 改成每 1000 行一次 SQL,SQL 往返从 ~3N 降到 ~5。 FactWriter 新增 bulkWrite(entries): - 1 次 SELECT 取所有相关 subject 的 latest version(by subjectId IN) - 内存里链式决策:unchanged / evidence_append / supersede / create - 处理 batch 内同 subject 多 draft(后 draft 跟前 draft 比,递推 liveLatest) - 1 个 $transaction commit:bulk updateMany supersede + bulk createMany 新版本 + 罕见 evidence_append 走 array_append raw SQL - {maxWait: 30s, timeout: 120s} 防大批量 + swap 下 5s 默认超时 ParserPipeline 新增 runForBatch(items): - 全部 tx 走 parser 收集 drafts(in-memory,无 DB) - 1 次 FactWriter.bulkWrite 提交;失败降级 per-entry writeDraft(保收尾) - 同 runForTransaction 的 metrics 接口,调用方零适配 cold-import processSubject 大重构: - 引入 buffer 按 tenant 分桶(因 createMany 不能跨 tenant) - 每 N 行触发 flushBatchedWrite: 1. createMany tx({skipDuplicates: true})— 1 SQL 写 N 行 2. SELECT WHERE sourceEventId IN (...) — 取回 tx ids 喂 parser 3. parserPipeline.runForBatch — 内部 bulkWrite - createMany 失败降级 fallbackPerRowWrite(per-row 老路径,保稳) - env PAC_WRITE_BATCH_SIZE 兜底(默认 1000;0 = 退回 per-row 回滚开关) 性能预期(实测待验证): per-row baseline: ~80 tx/s (实测服务器) bulk createMany + 10-20x → 800-1600 tx/s 4.6M 行全量: 9-12h → **30-60min** 跟 PR3 统一 sync 模式协同: - 任何 mode(sync/--full/cron 增量)都走同一条 hot path - cohort batch(PR2)+ write batch(PR4)正交叠加 - 失败降级保稳(createMany 崩 → fallback per-row;bulkWrite 崩 → fallback writeDraft) - 同 fact subject_id 跨 batch 一致性靠 version + partial UNIQUE active 兜底,不变 未来 PR5(可选):pg-copy-streams 真 COPY + staging 表 → 再 3-5x(总 30-50x)luoqi committed
| Name |
Last commit
|
Last update |
|---|---|---|
| .. | ||
| cli | Loading commit data... | |
| common | Loading commit data... | |
| config | Loading commit data... | |
| modules | Loading commit data... | |
| openapi | Loading commit data... | |
| prisma | Loading commit data... | |
| queues | Loading commit data... | |
| redis | Loading commit data... | |
| app.module.ts | Loading commit data... | |
| health.controller.ts | Loading commit data... | |
| main.ts | Loading commit data... |