fix: backfill_affiliations 游标修复 + recent-first 回填

- 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 万行验证倒序 + 无崩溃
2026-08-10 23:17:47 +08:00
parent 42649d13b5
commit cb4cb0d178
1 changed files with 102 additions and 33 deletions
@@ -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            # 全量，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")