Index per-source tag refreshes for scalable bulk ingestion

README.md

@@ -121,7 +121,7 @@ python3 -B collect_peerread.py --cache /data/paper-index/raw/http-cache \
   --out /data/paper-index/collected
 ```
 
-数据任务以一次性 systemd 服务运行：`paper-atlas-arxiv-import`、`paper-atlas-pmlr-collect`、`paper-atlas-peerread-collect`；`paper-atlas-pmlr-verify` 用于全索引复核，`paper-atlas-conference-merge` 用于单写入合并。它们不是定时任务。主库只允许一个写入进程；会议采集先写独立 JSONL，待 arXiv 主批次结束后再合并；也可以使用 `merge_conferences.py` 在保存断点处暂停本项目的 arXiv unit、顺序合并会议清单并恢复。该操作应作为独立的一次性 systemd 任务运行，避免 SSH 断开影响恢复；不能同时启动多个合并任务。运行日志位于 `/data/paper-index/*-import.log`、`/data/paper-index/*-collect.log`。任何服务启动成功都不等于批次已完成，需检查批次状态、清单与完整性结果。
+数据任务以一次性 systemd 服务运行：`paper-atlas-arxiv-import`、`paper-atlas-pmlr-collect`、`paper-atlas-peerread-collect`；`paper-atlas-pmlr-verify` 用于全索引复核，`paper-atlas-conference-merge` 用于单写入合并。它们不是定时任务。主库只允许一个写入进程；会议采集先写独立 JSONL，待 arXiv 主批次结束后再合并；也可以使用 `merge_conferences.py` 在保存断点处暂停本项目的 arXiv unit、顺序合并会议清单并恢复。该操作应作为独立的一次性 systemd 任务运行，避免 SSH 断开影响恢复；不能同时启动多个合并任务。运行日志位于 `/data/paper-index/*-import.log`、`/data/paper-index/*-collect.log`。任何服务启动成功都不等于批次已完成，需检查批次状态、清单与完整性结果。标签按来源刷新使用 `paper_tags_source_evidence` 索引，避免每条论文都扫描整张标签表。运行中的 arXiv 可以通过同一协调脚本的 `--schema-only` 模式在断点更新索引后恢复；此模式要求已有 arXiv unit 正在运行。
 
 ## 长期维护与更新
 

merge_conferences.py

@@ -10,6 +10,7 @@ import subprocess
 from pathlib import Path
 
 from import_records import ingest_file
+from index_core import Store
 
 
 UNIT='paper-atlas-arxiv-import.service'
@@ -27,6 +28,8 @@ def merge(db,archive,files):
             raise ValueError('Missing collected file or manifest: '+filename)
     if not Path(archive).is_file():
         raise ValueError('Missing arXiv archive')
+    if not Path(db).is_file():
+        raise ValueError('Missing existing database')
     if property_value('ActiveState')!='active':
         raise RuntimeError('Expected the existing arXiv unit to be active; inspect it before proceeding')
     subprocess.run(['systemctl','stop',UNIT],check=True)
@@ -34,12 +37,17 @@ def merge(db,archive,files):
         with sqlite3.connect(db) as c:
             checkpoint=c.execute("SELECT records_written FROM ingest_runs WHERE source_id=(SELECT id FROM sources WHERE name='arxiv') ORDER BY id DESC LIMIT 1").fetchone()
             print('Saved arxiv checkpoint:',checkpoint,flush=True)
+        schema_store=Store(db)
+        schema_store.conn.close()
         for filename in files:
             ingest_file(db,filename)
         with sqlite3.connect(db) as c:
-            errors=c.execute('PRAGMA foreign_key_check').fetchall()
-            if errors:
-                raise RuntimeError('Foreign key violations after merge')
+            if files:
+                errors=c.execute('PRAGMA foreign_key_check').fetchall()
+                if errors:
+                    raise RuntimeError('Foreign key violations after merge')
+            else:
+                print('Tag refresh plan:',c.execute('EXPLAIN QUERY PLAN DELETE FROM paper_tags WHERE source_record_id=? AND evidence_key=?',(1,'arxiv:categories')).fetchall(),flush=True)
             print('Decision totals:',c.execute('SELECT status,count(*) FROM decisions GROUP BY status').fetchall(),flush=True)
     finally:
         if property_value('LoadState')=='loaded':
@@ -55,6 +63,9 @@ if __name__=='__main__':
     parser=argparse.ArgumentParser()
     parser.add_argument('--db',required=True)
     parser.add_argument('--archive',required=True)
-    parser.add_argument('files',nargs='+')
+    parser.add_argument('--schema-only',action='store_true',help='Apply idempotent schema/index updates at a saved checkpoint')
+    parser.add_argument('files',nargs='*')
     args=parser.parse_args()
+    if not args.files and not args.schema_only:
+        parser.error('Provide collected files or --schema-only')
     merge(args.db,args.archive,args.files)

schema.sql

@@ -63,6 +63,7 @@ CREATE TABLE IF NOT EXISTS paper_tags (
  source_record_id INTEGER REFERENCES source_records(id),
  evidence_key TEXT NOT NULL, PRIMARY KEY(paper_id,tag_id,evidence_key)
 );
+CREATE INDEX IF NOT EXISTS paper_tags_source_evidence ON paper_tags(source_record_id,evidence_key);
 CREATE TABLE IF NOT EXISTS match_candidates (
  left_paper_id INTEGER NOT NULL REFERENCES papers(id), right_paper_id INTEGER NOT NULL REFERENCES papers(id),
  method TEXT NOT NULL, score REAL, evidence_json TEXT NOT NULL,