[fix] bugs in incremental update

Author: Melmaphother

.gitignore

@@ -29,3 +29,5 @@ site/
 
 # Legacy
 archive/
+
+.codex

config.py

@@ -46,6 +46,7 @@
 QDRANT_PATH: str | None = os.getenv("QDRANT_PATH")  # Non-empty: use local embedded mode
 QDRANT_COLLECTION_NAME: str = "papers"
 QDRANT_PREFER_GRPC: bool = True
+QDRANT_TIMEOUT: float = float(os.getenv("QDRANT_TIMEOUT", "300"))
 
 # ── API Server ─────────────────────────────────────────────────────────
 API_HOST: str = "0.0.0.0"
@@ -74,3 +75,4 @@
 INGEST_BATCH_SIZE: int = 5000
 SQLITE_CACHE_SIZE: int = -1024 * 512  # 512 MB (more cache = faster SQLite/FTS5)
 SQLITE_MMAP_SIZE: int = 512 * 1024 * 1024  # 512 MB for FTS5 read performance
+SQLITE_BUSY_TIMEOUT_MS: int = int(os.getenv("SQLITE_BUSY_TIMEOUT_MS", "300000"))  # 5 min

core/citation/database.py

@@ -78,9 +78,19 @@ def get_connection(readonly: bool = False) -> sqlite3.Connection:
     Path(config.PAPERS_DB_PATH).parent.mkdir(parents=True, exist_ok=True)
     if readonly:
         uri = f"file:{config.PAPERS_DB_PATH}?mode=ro"
-        conn = sqlite3.connect(uri, uri=True)
+        conn = sqlite3.connect(
+            uri,
+            uri=True,
+            timeout=config.SQLITE_BUSY_TIMEOUT_MS / 1000,
+        )
     else:
-        conn = sqlite3.connect(config.PAPERS_DB_PATH)
+        conn = sqlite3.connect(
+            config.PAPERS_DB_PATH,
+            timeout=config.SQLITE_BUSY_TIMEOUT_MS / 1000,
+        )
+        conn.execute("PRAGMA journal_mode = WAL")
+        conn.execute("PRAGMA synchronous = NORMAL")
+    conn.execute(f"PRAGMA busy_timeout = {config.SQLITE_BUSY_TIMEOUT_MS}")
     conn.execute(f"PRAGMA cache_size = {config.SQLITE_CACHE_SIZE}")
     return conn
 
@@ -222,28 +232,59 @@ def delete_by_corpus_ids(conn: sqlite3.Connection, corpus_ids: list[int]) -> Non
     """
     if not corpus_ids:
         return
-    placeholders = ",".join("?" for _ in corpus_ids)
-    paper_ids_subquery = f"(SELECT paper_id FROM corpus_id_mapping WHERE corpus_id IN ({placeholders}))"
-    conn.execute(
-        f"DELETE FROM paper_fts_title WHERE paper_id IN {paper_ids_subquery}",
-        corpus_ids,
-    )
-    conn.execute(
-        f"DELETE FROM paper_fts_combined WHERE paper_id IN {paper_ids_subquery}",
-        corpus_ids,
-    )
-    conn.execute(
-        f"DELETE FROM arxiv_to_paper WHERE paper_id IN {paper_ids_subquery}",
-        corpus_ids,
-    )
-    conn.execute(
-        f"DELETE FROM paper_metadata WHERE corpus_id IN ({placeholders})",
-        corpus_ids,
-    )
-    conn.execute(
-        f"DELETE FROM corpus_id_mapping WHERE corpus_id IN ({placeholders})",
-        corpus_ids,
-    )
+    for i in range(0, len(corpus_ids), config.INGEST_BATCH_SIZE):
+        batch_ids = corpus_ids[i : i + config.INGEST_BATCH_SIZE]
+        placeholders = ",".join("?" for _ in batch_ids)
+        paper_ids_subquery = f"(SELECT paper_id FROM corpus_id_mapping WHERE corpus_id IN ({placeholders}))"
+        conn.execute(
+            f"DELETE FROM paper_fts_title WHERE paper_id IN {paper_ids_subquery}",
+            batch_ids,
+        )
+        conn.execute(
+            f"DELETE FROM paper_fts_combined WHERE paper_id IN {paper_ids_subquery}",
+            batch_ids,
+        )
+        conn.execute(
+            f"DELETE FROM arxiv_to_paper WHERE paper_id IN {paper_ids_subquery}",
+            batch_ids,
+        )
+        conn.execute(
+            f"DELETE FROM paper_metadata WHERE corpus_id IN ({placeholders})",
+            batch_ids,
+        )
+        conn.execute(
+            f"DELETE FROM corpus_id_mapping WHERE corpus_id IN ({placeholders})",
+            batch_ids,
+        )
+
+
+def delete_by_paper_ids(conn: sqlite3.Connection, paper_ids: list[str]) -> None:
+    """Delete paper_metadata, corpus_id_mapping, arxiv_to_paper, and FTS tables by paper_id."""
+    if not paper_ids:
+        return
+    for i in range(0, len(paper_ids), config.INGEST_BATCH_SIZE):
+        batch_ids = paper_ids[i : i + config.INGEST_BATCH_SIZE]
+        placeholders = ",".join("?" for _ in batch_ids)
+        conn.execute(
+            f"DELETE FROM paper_fts_title WHERE paper_id IN ({placeholders})",
+            batch_ids,
+        )
+        conn.execute(
+            f"DELETE FROM paper_fts_combined WHERE paper_id IN ({placeholders})",
+            batch_ids,
+        )
+        conn.execute(
+            f"DELETE FROM arxiv_to_paper WHERE paper_id IN ({placeholders})",
+            batch_ids,
+        )
+        conn.execute(
+            f"DELETE FROM paper_metadata WHERE paper_id IN ({placeholders})",
+            batch_ids,
+        )
+        conn.execute(
+            f"DELETE FROM corpus_id_mapping WHERE paper_id IN ({placeholders})",
+            batch_ids,
+        )
 
 
 def delete_citations_by_ids(conn: sqlite3.Connection, citation_ids: list[int]) -> None:

core/db_pool.py

@@ -24,7 +24,7 @@ def _create_ro_connection() -> sqlite3.Connection:
     conn = sqlite3.connect(uri, uri=True, check_same_thread=False)
     conn.execute("PRAGMA journal_mode = WAL")
     conn.execute("PRAGMA wal_autocheckpoint = 0")
-    conn.execute("PRAGMA busy_timeout = 5000")
+    conn.execute(f"PRAGMA busy_timeout = {config.SQLITE_BUSY_TIMEOUT_MS}")
     conn.execute(f"PRAGMA cache_size = {config.SQLITE_CACHE_SIZE}")
     conn.execute(f"PRAGMA mmap_size = {config.SQLITE_MMAP_SIZE}")
     conn.execute("PRAGMA temp_store = MEMORY")

core/retrieve/dense.py

@@ -14,6 +14,7 @@
     Filter,
     FilterSelector,
     MatchAny,
+    PointIdsList,
     PointStruct,
     VectorParams,
 )
@@ -46,6 +47,7 @@ def get_client() -> QdrantClient:
                 host=config.QDRANT_HOST,
                 port=config.QDRANT_PORT,
                 prefer_grpc=config.QDRANT_PREFER_GRPC,
+                timeout=config.QDRANT_TIMEOUT,
             )
             print(f"✅ Qdrant connected: {config.QDRANT_HOST}:{config.QDRANT_PORT}")
     return _client
@@ -169,16 +171,8 @@ def delete_vectors(paper_ids: list[str]) -> None:
     if not paper_ids:
         return
     client = get_client()
+    point_ids = [_paper_id_to_point_id(pid) for pid in paper_ids]
     client.delete(
         collection_name=config.QDRANT_COLLECTION_NAME,
-        points_selector=FilterSelector(
-            filter=Filter(
-                must=[
-                    FieldCondition(
-                        key="paper_id",
-                        match=MatchAny(any=paper_ids),
-                    )
-                ]
-            )
-        ),
+        points_selector=PointIdsList(points=point_ids),
     )

docs/incremental-update.md

@@ -4,24 +4,42 @@
 
 ---
 
-## 1. 一键更新（推荐）
+## 1. 执行方式
 
 ```bash
-# 更新到最新 S2 release
+# 方式 A：一键串联（下载 + 校验 + merge）
 bash update.sh
 
-# 更新到指定日期
+# 更新到指定目标 release
 bash update.sh 2026-03-10
+
+# 方式 B：拆分执行
+# 1) 只下载增量 diff
+bash update_download.sh 2026-03-10
+
+# 2) 只校验已下载好的增量 diff
+bash update_validate.sh 2026-03-10
+
+# 3) 只 merge 已下载好的增量目录
+bash update_merge.sh PaperData/incremental/2026-01-27_to_2026-03-10
 ```
 
-脚本自动完成以下流程：
+推荐做法：
+
+1. 网络不稳定、只想先把文件拉下来时，先运行 `update_download.sh`
+2. 文件下完后，再运行 `update_validate.sh` 做完整校验与坏文件重下
+3. 需要重复调试 merge、断点恢复或只想继续上次合并时，直接运行 `update_merge.sh`
+4. `update.sh` 仅作为方便的一键封装，本质上顺序调用三者
+
+`update.sh` 自动完成以下流程：
 
 1. 从 `corpus/current_release.txt` 读取当前版本
 2. 查询 S2 API 获取可用 releases，确定目标版本
 3. 下载增量 diff 到 `PaperData/incremental/`
-4. 合并到 SQLite（paper_metadata、citations、ID 映射）+ FTS5
-5. 编码新增/修改论文的 BGE-M3 向量，upsert 到 Qdrant；删除已移除论文的向量
-6. 更新 `corpus/current_release.txt` 为最新版本
+4. 校验已下载 diff，并对损坏/缺失文件重下
+5. 合并到 SQLite（paper_metadata、citations、ID 映射）+ FTS5
+6. 编码新增/修改论文的 BGE-M3 向量，upsert 到 Qdrant；删除已移除论文的向量
+7. 更新 `corpus/current_release.txt` 为最新版本
 
 ### 前置条件
 
@@ -82,7 +100,8 @@ GET /diffs/{start_release_id}/to/{end_release_id}/{dataset_name}
 |-------------|--------------|
 | papers      | `corpusid`   |
 | abstracts   | `corpusid`   |
-| paper-ids   | `corpusid`   |
+| paper-ids updates  | `corpusid` |
+| paper-ids deletes  | `sha`      |
 | authors     | `authorid`   |
 | citations   | `citationid` |
 
@@ -95,13 +114,33 @@ GET /diffs/{start_release_id}/to/{end_release_id}/{dataset_name}
 ### 4.1 下载增量 diff
 
 ```bash
-# 自动从 corpus/current_release.txt 读取起始版本，下载到最新
-python build_corpus/data/download_incremental_diffs.py
+# 推荐：通过 shell 入口下载
+bash update_download.sh
+
+# 指定目标 release
+bash update_download.sh 2026-03-10
 
-# 指定起止版本
+# 单独校验并为损坏文件补下
+bash update_validate.sh 2026-03-10
+
+# 或直接调用 Python 脚本
+python build_corpus/data/download_incremental_diffs.py
 python build_corpus/data/download_incremental_diffs.py --start 2026-01-27 --end 2026-03-10
+python build_corpus/data/download_incremental_diffs.py --start 2026-01-27 --end 2026-03-10 --mode validate
 ```
 
+下载阶段与校验阶段现在分开：
+
+- `update_download.sh`
+  - 只负责“文件是否存在”
+  - 若目标文件已存在，直接跳过
+  - 若上次中断留下 `.tmp`，会自动断点续下
+- `update_validate.sh`
+  - 负责完整校验 `.gz/.jsonl` 内容
+  - 若文件缺失或损坏，会自动重新下载
+  - 会在增量目录下维护 `_download_validation_progress.json`
+  - 已校验通过的文件会被记录，下次校验时直接跳过
+
 ### 输出目录结构
 
 ```
@@ -130,6 +169,10 @@ PaperData/
 ### 4.2 合并到数据库
 
 ```bash
+# 推荐：通过 shell 入口 merge
+bash update_merge.sh PaperData/incremental/2026-01-27_to_2026-02-24
+
+# 或直接调用 Python 脚本
 python build_corpus/merge_incremental.py PaperData/incremental/2026-01-27_to_2026-02-24
 ```
 
@@ -139,6 +182,17 @@ python build_corpus/merge_incremental.py PaperData/incremental/2026-01-27_to_202
 - FTS5：刷新 paper_fts_title 和 paper_fts_combined
 - Qdrant：编码新增/修改论文的 BGE-M3 向量并 upsert，删除已移除论文的向量
 
+补充说明：
+
+- `merge_incremental.py` 会为每个数据集步骤显示文件级 `tqdm` 进度条
+- Qdrant 编码阶段也会显示批量进度
+- merge 过程中会在增量目录下写入 `_merge_progress.json`
+- `_merge_progress.json` 记录步骤级状态；对 `papers-updates`、`abstracts-updates`、`citations-updates`、`citations-deletes` 这类重步骤，会额外记录已成功 commit 的 chunk offset
+- 如果进程中断，重新执行同一 `INCR_DIR` 时会自动跳过已完成步骤并继续
+- 若上述重步骤中途失败，会从上一个成功 commit 的 chunk 继续，而不是从步骤开头完全重跑
+- 全部完成后 `_merge_progress.json` 会自动删除
+- 如果 merge 时发现损坏或不可读的 diff 文件，会直接报错并中止；应先回到下载阶段重新拉取文件
+
 ### 4.3 更新版本记录
 
 ```bash
@@ -151,9 +205,12 @@ echo "2026-02-24" > corpus/current_release.txt
 
 | 脚本 | 作用 |
 |------|------|
-| `update.sh` | 一键更新入口（下载 + 合并 + 版本更新） |
+| `update.sh` | 一键更新入口（调用下载脚本 + merge 脚本） |
+| `update_download.sh` | 只下载增量 diff，不执行完整校验、不执行 merge |
+| `update_validate.sh` | 只校验增量 diff，并对缺失/损坏文件重下 |
+| `update_merge.sh` | 只对指定增量目录执行 merge，并更新版本记录 |
 | `build_corpus/data/download_incremental_diffs.py` | 下载增量 diff 到 PaperData 目录 |
-| `build_corpus/merge_incremental.py` | 合并增量到 SQLite + FTS5 + Qdrant |
+| `build_corpus/merge_incremental.py` | 合并增量到 SQLite + FTS5 + Qdrant，支持断点续传 |
 | `build_corpus/optimize_fts.py` | FTS5 索引优化（建议更新后执行） |
 
 ---
@@ -163,5 +220,11 @@ echo "2026-02-24" > corpus/current_release.txt
 - 需要配置 `S2_API_KEY`（`.env` 或环境变量）以访问 Datasets API。
 - diff 的 `update_files`、`delete_files` 为预签名 URL，有时效，建议下载后本地保存。
 - 合并时需保证主键一致（如 `corpusid`、`citationid`、`authorid`）。
+- `paper-ids` 的主键口径需要分开看：`updates` 用 `corpusid`，`deletes` 用 `sha`
 - 合并完成后增量文件保留在 `PaperData/incremental/` 中，不会自动删除。
 - 合并过程中 Qdrant 向量编码需要 GPU，确保有可用的 CUDA 设备。
+- 如果只想补文件，不必每次完整校验；先跑 `update_download.sh` 即可。
+- 如果只想重复 merge，不要先跑下载/校验；直接执行 `update_merge.sh` 即可。
+- `_merge_progress.json` 属于 merge 的运行时状态文件；若一次 merge 已完整成功，它会自动被清理。
+- `_download_validation_progress.json` 属于校验阶段的运行时状态文件；会保留在增量目录下，供下次继续校验时复用。
+- chunk 或步骤重跑不会在 SQLite 或 Qdrant 中产生“重复数据行”；当前实现依赖 upsert/delete 的幂等性，但会产生额外耗时。