diff --git a/docs/operations.md b/docs/operations.md index a7cf4d7..ae875f2 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -123,3 +123,36 @@ paper-atlas maintain pmlr-tracks --db /data/paper-index/papers.sqlite \ 检查报告后，在同一命令添加 `--apply` 并使用新的 `--out` 路径执行迁移。计划使用共享维护锁，实际迁移使用独占锁；已有 arXiv 类别修复或其他写入运行时会拒绝启动，不会停止或重启它们。一次事务内核对整个来源集合、原始题目/摘要/作者、forum 身份、已有事件及目标冲突，更新原投稿的 edition 和身份归属，回填官方轨道出处及逐轨道覆盖数，最后复查轨道与外键后提交。论文、来源、投稿、决定的 ID 及决定观察均保留，不重新创建事件、不做实体合并、不改论文正文元数据或标签。 报告记录每个迁移事件的旧 edition、原身份、完整决定观察及新轨道证据。`planned` 是未执行的计划；`complete` 才表示事务已提交；`error` 表示执行失败。若进程在提交前后中断而报告停在 `prepared`，先对照主库和报告核查，不根据日志缺失自行重启任务。新路径重跑会重新核对当前数据库，已正确的轨道不会再次迁移。共享多个来源、重复目标事件或冲突决定会停止迁移，需另行核对，不能直接删掉其中一条。 + +## 追加同一目录的部分记录 + +默认 `import records` 把 manifest 当成一份来源快照，覆盖统计使用该文件的数量。 +若目录已有2条，之后仅补入1条，不能把这一小批当成完整快照，否则会把目录计数改成1。 +为已核验的部分补充批次，在 manifest 显式声明： + +```json +{ + "coverage_mode": "incremental", + "coverage_previous_count": 2, + "coverage_previous_expected_count": 2, + "coverage_expected_count": 4, + "status": "partial", + "count": 1 +} +``` + +这只是新增字段示例，manifest仍须保留source、venue、year、track、list_url、文件哈希等来源证据。 +`count`是本文件行数，`coverage_previous_count`是导入前该来源/届次的实际记录数， +`coverage_previous_expected_count`是旧覆盖行的expected_count（未知时用null）， +`coverage_expected_count`是独立目录证据支持的整份来源预期数，不能小于已知旧预期数。 +这个例子成功后覆盖统计为预期4、已导入3；文件checkpoint的records_written仍为1。 +刷新相同记录不重复累加。初次导入部分来源可声明previous_count为0、previous_expected_count为null。 + +只支持当前最多一个规范目录URL且旧覆盖状态为partial的来源/届次；多目录、旧状态complete、 +旧计数与实际来源记录不符，或前置计数已变化，都会中止，先核对来源范围或重新取得完整快照。 +该模式在写入前检查父状态，并在同一文件事务中按实际来源记录重新计数、检查总量界限和乐观更新覆盖行。 +失败会回滚本文件的论文、来源、决定、标签和覆盖变化，保留失败运行记录；更早已提交文件不重放。 +增量模式始终保留partial，不能靠一个小批次推断完整论文集、摘要或公开拒稿覆盖。 + +仍需先独立核对新增记录身份与决定，按原生单写入保护串行导入，再独立核验数据与原有历史。 +这些计数检查不证明外部目录完整，也不代替最终全库去重。 diff --git a/paper_atlas/ingestion/records.py b/paper_atlas/ingestion/records.py index b303ce8..60f1cb2 100644 --- a/paper_atlas/ingestion/records.py +++ b/paper_atlas/ingestion/records.py @@ -12,6 +12,55 @@ from paper_atlas.collectors.public_openreview import SOURCE as PUBLIC_SOURCE from paper_atlas.collectors.icml_public_archive import SOURCE as ARCHIVE_SOURCE +def coverage_options(manifest): + mode = manifest.get('coverage_mode', 'snapshot') + if mode not in ('snapshot', 'incremental'): + raise ValueError('Unsupported coverage mode') + if mode == 'snapshot': + return None + required = ('venue', 'year', 'track', 'count', 'coverage_previous_count', + 'coverage_previous_expected_count', 'coverage_expected_count') + if any(key not in manifest for key in required) or manifest.get('status') != 'partial': + raise ValueError('Incremental coverage requires a partial, explicitly scoped manifest') + previous = manifest['coverage_previous_count'] + expected = manifest['coverage_expected_count'] + prior_expected = manifest['coverage_previous_expected_count'] + if (type(previous) is not int or previous < 0 or type(expected) is not int or expected < 1 + or type(manifest['count']) is not int or manifest['count'] < 1 + or expected < max(previous, manifest['count']) + or (prior_expected is not None and (type(prior_expected) is not int or prior_expected < 0 + or expected < prior_expected))): + raise ValueError('Invalid incremental coverage counts') + return previous, prior_expected, expected + + +def source_record_count(c, source, edition): + return c.execute('''SELECT count(DISTINCT ss.source_record_id) FROM submission_sources ss + JOIN submissions s ON s.id=ss.submission_id + JOIN source_records sr ON sr.id=ss.source_record_id + WHERE s.edition_id=? AND sr.source_id=?''', (edition, source)).fetchone()[0] + + +def require_incremental_parent(c, source, manifest, options): + edition = c.execute('''SELECT e.id FROM editions e JOIN venues v ON v.id=e.venue_id + WHERE lower(v.acronym)=lower(?) AND e.year=? AND e.track=?''', + (manifest['venue'], manifest['year'], manifest['track'])).fetchone() + entries = c.execute('''SELECT list_url,status,expected_count,imported_count + FROM edition_sources WHERE edition_id=? AND source_id=?''', + (edition[0], source)).fetchall() if edition else [] + previous, prior_expected, _ = options + if entries: + if len(entries) != 1 or entries[0][0] != manifest['list_url']: + raise ValueError('Incremental coverage requires one unambiguous source listing') + if tuple(entries[0][1:]) != ('partial', prior_expected, previous): + raise ValueError('Incremental coverage parent changed or is not partial') + elif previous != 0 or prior_expected is not None: + raise ValueError('Incremental coverage parent is absent') + actual = source_record_count(c, source, edition[0]) if edition else 0 + if actual != previous: + raise ValueError('Incremental coverage parent differs from actual source records') + + def ingest_file(db,path, *, pg_service=None, pg_schema=None): path = Path(path) manifest_path = path.with_suffix('.manifest.json') @@ -24,6 +73,7 @@ def ingest_file(db,path, *, pg_service=None, pg_schema=None): source_name = manifest.get('source') if not source_name: raise ValueError('A source manifest is required') + incremental = coverage_options(manifest) store = open_store(db, pg_service=pg_service, pg_schema=pg_schema) c = store.conn run = None @@ -35,6 +85,8 @@ def ingest_file(db,path, *, pg_service=None, pg_schema=None): source_id = store.source(source_name,manifest['list_url']) run = inserted_id(c,'INSERT INTO ingest_runs(source_id,checkpoint_json) VALUES(?,?)',(source_id,json.dumps({'file':str(path),'expected_sha256':manifest.get('file_sha256')}))) c.commit() + if incremental is not None: + require_incremental_parent(c, source_id, manifest, incremental) digest = hashlib.sha256() with path.open('rb') as stream: for line in stream: @@ -82,8 +134,25 @@ def ingest_file(db,path, *, pg_service=None, pg_schema=None): if expected is not None and expected != count: raise ValueError(f'Manifest count {expected} differs from records {count}') complete=expected==count and manifest.get('status') not in ('partial','blocked') and not manifest.get('missing_abstract_count',0) and not missing_abstracts - c.execute('INSERT INTO edition_sources(edition_id,source_id,list_url,status,expected_count,imported_count,last_success_at) VALUES(?,?,?,?,?,?,?) ON CONFLICT(edition_id,source_id,list_url) DO UPDATE SET status=excluded.status,expected_count=excluded.expected_count,imported_count=excluded.imported_count,last_success_at=excluded.last_success_at,last_error=NULL',(edition,source_id,manifest['list_url'],'complete' if complete else 'partial',expected,count,time.strftime('%Y-%m-%dT%H:%M:%SZ',time.gmtime()))) - c.execute("UPDATE ingest_runs SET status='complete',finished_at=?,records_seen=?,records_written=?,checkpoint_json=? WHERE id=?",(timestamp(),count,count,json.dumps({'file':str(path),'file_sha256':digest.hexdigest()}),run)) + imported = count + condition = '' + condition_params = () + if incremental is not None: + previous, prior_expected, expected = incremental + imported = source_record_count(c, source_id, edition) + if imported < previous or imported > expected: + raise ValueError('Incremental coverage exceeds the declared source bounds') + complete = False + condition = " WHERE edition_sources.imported_count=? AND COALESCE(edition_sources.expected_count,-1)=? AND edition_sources.status='partial'" + condition_params = (previous, -1 if prior_expected is None else prior_expected) + result = c.execute('INSERT INTO edition_sources(edition_id,source_id,list_url,status,expected_count,imported_count,last_success_at) VALUES(?,?,?,?,?,?,?) ON CONFLICT(edition_id,source_id,list_url) DO UPDATE SET status=excluded.status,expected_count=excluded.expected_count,imported_count=excluded.imported_count,last_success_at=excluded.last_success_at,last_error=NULL'+condition,(edition,source_id,manifest['list_url'],'complete' if complete else 'partial',expected,imported,time.strftime('%Y-%m-%dT%H:%M:%SZ',time.gmtime()),*condition_params)) + if incremental is not None and result.rowcount != 1: + raise ValueError('Incremental coverage parent changed during import') + checkpoint = {'file':str(path),'file_sha256':digest.hexdigest()} + if incremental is not None: + checkpoint['coverage'] = {'mode':'incremental','previous_count':previous, + 'expected_count':expected,'imported_count':imported} + c.execute("UPDATE ingest_runs SET status='complete',finished_at=?,records_seen=?,records_written=?,checkpoint_json=? WHERE id=?",(timestamp(),count,count,json.dumps(checkpoint),run)) c.commit() print(json.dumps({'file':str(path),'status':'imported','records':count,'missing_abstract_count':missing_abstracts}),flush=True) except BaseException as exc: diff --git a/tests/test_incremental_coverage.py b/tests/test_incremental_coverage.py new file mode 100644 index 0000000..6781b9e --- /dev/null +++ b/tests/test_incremental_coverage.py @@ -0,0 +1,116 @@ +from contextlib import redirect_stdout +import hashlib +import io +import json +from pathlib import Path +import sqlite3 +import tempfile +import unittest + +from paper_atlas.ingestion.records import ingest_file + + +class IncrementalCoverageTests(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.addCleanup(self.tmp.cleanup) + self.root = Path(self.tmp.name) + self.db = self.root / 'papers.sqlite' + + def file(self, name, external_ids, **options): + rows = [{'source':'example','external_id':key,'url':'https://example.org/'+key, + 'venue':'Example','year':2026,'track':'main','status':'accepted', + 'title':'Independent research contribution '+key,'abstract':'Known abstract', + 'authors':[], 'identifiers':{'doi':'10.1234/'+key}} for key in external_ids] + path = self.root / (name+'.jsonl') + path.write_bytes((''.join(json.dumps(row)+'\n' for row in rows)).encode()) + manifest = {key:rows[0][key] for key in ('source','venue','year','track')} + manifest.update(list_url='https://example.org/2026', count=len(rows),status='partial', + file_sha256=hashlib.sha256(path.read_bytes()).hexdigest()) + manifest.update(options) + path.with_suffix('.manifest.json').write_text(json.dumps(manifest)) + return path + + def ingest(self, path): + with redirect_stdout(io.StringIO()): + ingest_file(self.db,path) + + def incremental(self, previous=2, prior_expected=2, expected=4): + return {'coverage_mode':'incremental','coverage_previous_count':previous, + 'coverage_previous_expected_count':prior_expected,'coverage_expected_count':expected} + + def coverage(self): + with sqlite3.connect(self.db) as c: + return c.execute('SELECT status,expected_count,imported_count FROM edition_sources').fetchall() + + def test_append_and_refresh_count_actual_source_records_without_erasing_earlier_batch(self): + self.ingest(self.file('first',['a','b'])) + self.ingest(self.file('append',['c'],**self.incremental())) + self.assertEqual(self.coverage(),[('partial',4,3)]) + self.ingest(self.file('refresh',['c'],**self.incremental(3,4,4))) + self.assertEqual(self.coverage(),[('partial',4,3)]) + with sqlite3.connect(self.db) as c: + self.assertEqual(c.execute('SELECT count(*) FROM source_records').fetchone()[0],3) + self.assertEqual(c.execute('SELECT count(*) FROM submissions').fetchone()[0],3) + runs = c.execute('SELECT records_written,checkpoint_json FROM ingest_runs ORDER BY id').fetchall() + self.assertEqual([r[0] for r in runs],[2,1,1]) + self.assertEqual(json.loads(runs[-1][1])['coverage'], + {'mode':'incremental','previous_count':3,'expected_count':4,'imported_count':3}) + self.assertEqual(c.execute('PRAGMA foreign_key_check').fetchall(),[]) + + def test_stale_parent_stops_before_source_writes(self): + self.ingest(self.file('first',['a','b'])) + with self.assertRaisesRegex(ValueError,'parent changed'): + self.ingest(self.file('stale',['c'],**self.incremental(1,2,4))) + self.assertEqual(self.coverage(),[('partial',2,2)]) + with sqlite3.connect(self.db) as c: + self.assertEqual(c.execute("SELECT count(*) FROM source_records WHERE external_id='c'").fetchone()[0],0) + self.assertEqual(c.execute('SELECT status FROM ingest_runs ORDER BY id DESC LIMIT 1').fetchone()[0],'failed') + + def test_late_source_bound_failure_rolls_back_new_records_and_coverage(self): + self.ingest(self.file('first',['a','b'])) + with self.assertRaisesRegex(ValueError,'source bounds'): + self.ingest(self.file('overflow',['c'],**self.incremental(2,2,2))) + self.assertEqual(self.coverage(),[('partial',2,2)]) + with sqlite3.connect(self.db) as c: + for table in ('papers','source_records','submissions','decisions'): + self.assertEqual(c.execute('SELECT count(*) FROM '+table).fetchone()[0],2) + + def test_multiple_listings_cannot_be_silently_aggregated(self): + self.ingest(self.file('one',['a'])) + self.ingest(self.file('two',['b'],list_url='https://example.org/another-list')) + before=self.coverage() + with self.assertRaisesRegex(ValueError,'unambiguous source listing'): + self.ingest(self.file('ambiguous',['c'],**self.incremental(2,1,4))) + self.assertEqual(self.coverage(),before) + + def test_untracked_source_rows_require_reconciliation(self): + self.ingest(self.file('first',['a','b'])) + with sqlite3.connect(self.db) as c: + c.execute('UPDATE edition_sources SET imported_count=1') + with self.assertRaisesRegex(ValueError,'actual source records'): + self.ingest(self.file('stale-summary',['c'],**self.incremental(1,2,4))) + self.assertEqual(self.coverage(),[('partial',2,1)]) + + def test_first_partial_chunk_can_declare_an_independently_known_source_total(self): + self.ingest(self.file('first',['a'],**self.incremental(0,None,4))) + self.assertEqual(self.coverage(),[('partial',4,1)]) + + def test_complete_listing_is_not_demoted_by_a_partial_chunk(self): + self.ingest(self.file('complete',['a','b'],status='complete')) + with self.assertRaisesRegex(ValueError,'not partial'): + self.ingest(self.file('append',['c'],**self.incremental())) + self.assertEqual(self.coverage(),[('complete',2,2)]) + + def test_invalid_mode_counts_and_missing_scope_fail_before_database_creation(self): + for options in ({'coverage_mode':'unknown'}, self.incremental(True,2,4), + self.incremental(2,5,4),self.incremental(-1,2,4), + {'coverage_mode':'incremental'}, + {**self.incremental(),'status':'complete'}): + with self.subTest(options=options): + with self.assertRaises(ValueError):self.ingest(self.file('invalid',['a'],**options)) + self.assertFalse(self.db.exists()) + + +if __name__=='__main__': + unittest.main() diff --git a/tests/test_postgres_ingestion.py b/tests/test_postgres_ingestion.py index a37c5fa..95aaf55 100644 --- a/tests/test_postgres_ingestion.py +++ b/tests/test_postgres_ingestion.py @@ -87,6 +87,82 @@ class PostgreSQLIngestionTests(unittest.TestCase): with redirect_stdout(io.StringIO()): ingest_file(None, path, pg_service=self.service, pg_schema=self.schema) + def test_incremental_source_coverage_counts_prior_chunks_and_refreshes_on_postgres(self): + rows = [self.row(source='dblp',external=key,status='accepted',year=2001,venue='UAI') + for key in ('incremental1','incremental2','incremental3')] + listing = 'https://example.org/uai/2001' + def manifest(path, **options): + target=path.with_suffix('.manifest.json') + value=json.loads(target.read_text()) + value.update(list_url=listing,status='partial',**options) + target.write_text(json.dumps(value)) + first=self.file('initial',rows[:2]);manifest(first);self.ingest(first) + second=self.file('append',[rows[2]]) + manifest(second,coverage_mode='incremental',coverage_previous_count=2, + coverage_previous_expected_count=2,coverage_expected_count=4) + self.ingest(second) + manifest(second,coverage_mode='incremental',coverage_previous_count=3, + coverage_previous_expected_count=4,coverage_expected_count=4) + self.ingest(second) + with closing(ReadConnection(self.service,self.schema)) as c: + coverage=c.execute('SELECT status,expected_count,imported_count FROM edition_sources WHERE list_url=?',(listing,)).fetchone() + self.assertEqual(tuple(coverage.values),('partial',4,3)) + stored=c.execute('SELECT abstract FROM source_records WHERE external_id=?',('incremental3',)).fetchone() + self.assertEqual(stored[0],rows[2]['abstract']) + self.assertEqual(c.execute("SELECT count(*) FROM source_records r JOIN sources s ON s.id=r.source_id WHERE s.name='dblp'").fetchone()[0],3) + run=c.execute('SELECT records_written,checkpoint_json FROM ingest_runs ORDER BY id DESC LIMIT 1').fetchone() + self.assertEqual(run[0],1) + self.assertEqual(json.loads(run[1])['coverage']['imported_count'],3) + + def test_incremental_postgres_bound_failure_rolls_back_rows_and_summary(self): + listing='https://example.org/uai/2001' + rows=[self.row(source='dblp',external=key,status='accepted',year=2001,venue='UAI') + for key in ('bound1','bound2','bound3')] + first=self.file('initial',rows[:2]);target=first.with_suffix('.manifest.json') + value=json.loads(target.read_text());value.update(list_url=listing,status='partial');target.write_text(json.dumps(value)) + self.ingest(first) + with closing(ReadConnection(self.service,self.schema)) as c: + before={table:c.execute('SELECT count(*) FROM '+table).fetchone()[0] + for table in ('papers','source_records','submissions','decisions')} + path=self.file('overflow',[rows[2]]);target=path.with_suffix('.manifest.json') + value=json.loads(target.read_text());value.update(list_url=listing,status='partial',coverage_mode='incremental', + coverage_previous_count=2,coverage_previous_expected_count=2,coverage_expected_count=2) + target.write_text(json.dumps(value)) + with self.assertRaisesRegex(ValueError,'source bounds'):self.ingest(path) + with closing(ReadConnection(self.service,self.schema)) as c: + self.assertEqual({table:c.execute('SELECT count(*) FROM '+table).fetchone()[0] for table in before},before) + coverage=c.execute('SELECT status,expected_count,imported_count FROM edition_sources WHERE list_url=?',(listing,)).fetchone() + self.assertEqual(tuple(coverage.values),('partial',2,2)) + self.assertEqual(c.execute('SELECT status FROM ingest_runs ORDER BY id DESC LIMIT 1').fetchone()[0],'failed') + + def test_incremental_postgres_does_not_overwrite_a_concurrently_changed_parent(self): + listing='https://example.org/uai/2001' + rows=[self.row(source='dblp',external=key,status='accepted',year=2001,venue='UAI') + for key in ('concurrent1','concurrent2','concurrent3')] + first=self.file('initial',rows[:2]);target=first.with_suffix('.manifest.json') + value=json.loads(target.read_text());value.update(list_url=listing,status='partial');target.write_text(json.dumps(value)) + self.ingest(first) + path=self.file('racing',[rows[2]]);target=path.with_suffix('.manifest.json') + value=json.loads(target.read_text());value.update(list_url=listing,status='partial',coverage_mode='incremental', + coverage_previous_count=2,coverage_previous_expected_count=2,coverage_expected_count=4) + target.write_text(json.dumps(value)) + original=Store.record + def change_parent(store,*args,**kwargs): + result=original(store,*args,**kwargs) + # A separate committed SQL writer need not honor the native + # advisory lock. Its update must survive our rejected file. + with connect(self.service) as other: + other.execute('SET search_path TO '+quote(self.schema)+',pg_catalog') + other.execute('UPDATE edition_sources SET expected_count=99 WHERE list_url=%s',(listing,)) + return result + with patch.object(Store,'record',change_parent): + with self.assertRaisesRegex(ValueError,'parent changed during import'):self.ingest(path) + with closing(ReadConnection(self.service,self.schema)) as c: + coverage=c.execute('SELECT status,expected_count,imported_count FROM edition_sources WHERE list_url=?',(listing,)).fetchone() + self.assertEqual(tuple(coverage.values),('partial',99,2)) + self.assertEqual(c.execute('SELECT count(*) FROM source_records WHERE external_id=?',('concurrent3',)).fetchone()[0],0) + self.assertEqual(c.execute('SELECT status FROM ingest_runs ORDER BY id DESC LIMIT 1').fetchone()[0],'failed') + def test_native_writes_keep_nul_historical_ids_and_later_acceptance(self): store = self.store() try: