From cb4cb0d17811d52738545cfb2f4cc8220dc7d9f1 Mon Sep 17 00:00:00 2001 From: "34047007@qq.com" <34047007@qq.com> Date: Mon, 10 Aug 2026 23:17:47 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20backfill=5Faffiliations=20=E6=B8=B8?= =?UTF-8?q?=E6=A0=87=E4=BF=AE=E5=A4=8D=20+=20recent-first=20=E5=9B=9E?= =?UTF-8?q?=E5=A1=AB?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - ctid 游标：asyncpg 的 tid 编解码器只接受 (page, offset) 元组，直接绑 'ctid::text' 字符串到 $n::tid 会报 "list or tuple expected"（容器 crash 根因）。 解析 'ctid::text' → 元组后绑定，本地真实库 6 万行验证通过 - 默认改 recent-first 游标：按 (pub_date DESC NULLS LAST, id DESC) keyset 推进 （走 ix_gl_pub_date_sort 索引，谓词复用 search_engine._keyset_condition 的 date 分支），最新文献先回填——affiliation 搜索对近期文献先恢复可用， 而非等全量 ~10 小时（表 FILLFACTOR=100 页满 → 非 HOT → 35 索引 fan-out， 单索引删除不改善，仅改善顺序不缩短总时长） - --physical 保留原 ctid 游标作回退；本地真实库 3 批 6 万行验证倒序 + 无崩溃 --- backend/scripts/backfill_affiliations.py | 135 +++++++++++++++++------ 1 file changed, 102 insertions(+), 33 deletions(-) diff --git a/backend/scripts/backfill_affiliations.py b/backend/scripts/backfill_affiliations.py index f5be5b6..fe9b1e2 100644 --- a/backend/scripts/backfill_affiliations.py +++ b/backend/scripts/backfill_affiliations.py @@ -4,18 +4,28 @@ AccessExclusiveLock 整表，阻塞搜索读）。本脚本分批后台执行： - 每批 2 万行，批间提交释放锁 -- ctid 游标（物理序）顺序推进，每行只读一次，避免 O(n²) 重扫 +- **默认 recent-first 游标**：按 (pub_date DESC NULLS LAST, id DESC) keyset 推进 + （走 ix_gl_pub_date_sort 索引），最新文献先回填——affiliation 搜索对近期文献 + 先恢复可用，而不是等全量 10 小时结束。keyset 谓词与 search_engine 的 + _keyset_condition(date) 分支一致（含 pub_date IS NULL 区域用 id 续推） +- `--physical` 走 ctid 物理序游标（旧路径，保留作回退） - 幂等：WHERE affiliations_text IS NULL，可随时重跑/中断续跑 -- 串行强制（max_parallel_workers_per_gather=0）：并行 SeqScan 不保 ctid 序， - max(ctid) 游标会漏行 +- 串行强制（max_parallel_workers_per_gather=0）：并行 SeqScan 不保序， + 物理/keyset 游标都可能漏行 - 新插入行由触发器直接维护，脚本只处理存量 - 每 _VACUUM_EVERY 批执行 VACUUM：非 HOT 更新会积累死元组（旧元组+36 索引旧条目 VACUUM 前不释放），全量不清理瞬时占用≈整表大小（27GB）会爆盘。批间 VACUUM 允许并发读写（不阻塞搜索），把瞬时峰值压到 ~1-2GB +⚠️ 速度现实（本地 578 万行实测 ~7.3ms/行，全量约 10 小时）：表 FILLFACTOR=100、 +页 ~99% 满，任何 UPDATE 新元组都放不进原页 → 放弃 HOT → 35 个索引（12 GIN）全写 +新 TID。这是 DDL 迁移外的固有成本，删除单个索引（如 trgm）不改善。recent-first +只改善"哪些先可用"，不缩短总时长。 + 用法: - cd backend && python scripts/backfill_affiliations.py # 全量 - cd backend && python scripts/backfill_affiliations.py 500000 # 最多处理 N 行后退出 + cd backend && python scripts/backfill_affiliations.py # 全量，recent-first + cd backend && python scripts/backfill_affiliations.py 500000 # 最多处理 N 行后退出 + cd backend && python scripts/backfill_affiliations.py --physical # 物理序游标（旧路径） """ import asyncio import sys @@ -29,6 +39,71 @@ from app.config import Settings _BATCH = 20_000 _VACUUM_EVERY = 10 # 每 10 批执行一次 VACUUM，回收死元组空间，防止瞬时膨胀爆盘 +_FILTER = "affiliations_text IS NULL AND authors IS NOT NULL AND authors != '[]'::jsonb" +_ORDER = "pub_date DESC NULLS LAST, id DESC" + + +def _parse_ctid(ctid_text: str) -> tuple[int, int]: + """'(892927,3)' → (892927, 3)。 + + asyncpg 的 tid 编解码器只接受 (page, offset) 二元组，直接绑 'ctid::text' + 字符串到 $n::tid 会报 "list or tuple expected"。比较时用 (page, offset) + 元组（ctid 支持 btree 比较），绕过 asyncpg 编解码限制。 + """ + page_s, offset_s = ctid_text.strip("()").split(",") + return (int(page_s), int(offset_s)) + + +async def _fetch_recent(conn: asyncpg.Connection, cursor, batch: int): + """recent-first 游标：按 (pub_date DESC NULLS LAST, id DESC) keyset 推进。 + + cursor 形态: + None → 起点（最新在前） + ("date", date, last_id) → 已到非 NULL 区域，按日期续推（id 做 tiebreaker） + ("null", uuid_str) → 已进入 pub_date IS NULL 区域，按 id 续推 + 谓词与 search_engine._keyset_condition 的 date 分支一致（P12 已验证）。 + """ + if cursor is None: + return await conn.fetch( + f"SELECT id, pub_date FROM global_literature " + f"WHERE {_FILTER} ORDER BY {_ORDER} LIMIT $1", + batch, + ) + kind = cursor[0] + if kind == "null": + return await conn.fetch( + f"SELECT id, pub_date FROM global_literature " + f"WHERE {_FILTER} AND pub_date IS NULL AND id < $1 " + f"ORDER BY {_ORDER} LIMIT $2", + cursor[1], + batch, + ) + _, val, last_id = cursor + return await conn.fetch( + f"SELECT id, pub_date FROM global_literature " + f"WHERE {_FILTER} AND (pub_date < $1 OR (pub_date = $1 AND id < $2) OR pub_date IS NULL) " + f"ORDER BY {_ORDER} LIMIT $3", + val, + last_id, + batch, + ) + + +async def _fetch_physical(conn: asyncpg.Connection, last, batch: int): + """ctid 物理序游标（--physical 回退路径）。""" + if last is None: + return await conn.fetch( + f"SELECT id, ctid::text AS ctid FROM global_literature " + f"WHERE {_FILTER} LIMIT $1", + batch, + ) + return await conn.fetch( + f"SELECT id, ctid::text AS ctid FROM global_literature " + f"WHERE ctid > $1::tid AND {_FILTER} LIMIT $2", + last, + batch, + ) + async def main() -> None: settings = Settings() @@ -39,47 +114,41 @@ async def main() -> None: try: await conn.execute("SET max_parallel_workers_per_gather = 0") remaining = await conn.fetchval( - "SELECT count(*) FROM global_literature WHERE affiliations_text IS NULL " - "AND authors IS NOT NULL AND authors != '[]'::jsonb" + f"SELECT count(*) FROM global_literature WHERE {_FILTER}" ) print(f"待回填: {remaining} 条") limit: int | None = None - if len(sys.argv) > 1: - limit = int(sys.argv[1]) + physical = False + for a in sys.argv[1:]: + if a == "--physical": + physical = True + elif a.isdigit(): + limit = int(a) + if limit is not None: print(f"本次最多处理 {limit} 条") + print(f"游标模式: {'physical(ctid)' if physical else 'recent-first(pub_date DESC)'}") - last: str | None = None + cursor = None # recent-first: (kind, val)；physical: ctid 元组 processed = 0 while True: if limit is not None and processed >= limit: print(f"达到上限 {limit}，退出") break - if last is None: - rows = await conn.fetch( - """ - SELECT id, ctid::text AS ctid FROM global_literature - WHERE affiliations_text IS NULL - AND authors IS NOT NULL AND authors != '[]'::jsonb - LIMIT $1 - """, - _BATCH, - ) + if physical: + rows = await _fetch_physical(conn, cursor, _BATCH) else: - rows = await conn.fetch( - """ - SELECT id, ctid::text AS ctid FROM global_literature - WHERE ctid > $1::tid - AND affiliations_text IS NULL - AND authors IS NOT NULL AND authors != '[]'::jsonb - LIMIT $2 - """, - last, - _BATCH, - ) + rows = await _fetch_recent(conn, cursor, _BATCH) if not rows: break ids = [r["id"] for r in rows] - last = rows[-1]["ctid"] + if physical: + cursor = _parse_ctid(rows[-1]["ctid"]) + else: + last_row = rows[-1] + if last_row["pub_date"] is None: + cursor = ("null", last_row["id"]) + else: + cursor = ("date", last_row["pub_date"], last_row["id"]) await conn.execute( """ UPDATE global_literature @@ -93,7 +162,7 @@ async def main() -> None: ids, ) processed += len(ids) - print(f"已处理 {processed} 行（last ctid {last}）", flush=True) + print(f"已处理 {processed} 行（last {cursor}）", flush=True) if processed % (_BATCH * _VACUUM_EVERY) == 0: # VACUUM 并发读写安全，只回收死元组（不阻塞搜索读），压住瞬时膨胀 await conn.execute("VACUUM global_literature")