From 9cc7b5b050c64c9e78a0ccc05fd2e4919ebdfac2 Mon Sep 17 00:00:00 2001 From: Codex Bot Date: Sun, 8 Nov 2026 15:47:07 +0800 Subject: [PATCH 1/3] Keep stable paper identities separate from ambiguous title matches --- docs/usage.md | 2 ++ paper_atlas/db/store.py | 20 ++++++++++----- tests/test_index_core.py | 44 ++++++++++++++++++++++++++++++++ tests/test_postgres_ingestion.py | 28 ++++++++++++++++++++ 4 files changed, 87 insertions(+), 7 deletions(-) diff --git a/docs/usage.md b/docs/usage.md index b849333..452f3d0 100644 --- a/docs/usage.md +++ b/docs/usage.md @@ -99,6 +99,8 @@ paper-atlas query --pg-service foundry-research \ `--stats` 中 papers 是当前规则下的论文实体数；source_records 是来源记录数；decisions 是决定观察数。一个投稿可能有多个来源和多条观察，不能相加当作论文数。 +已有稳定标识或原来源记录匹配时，该身份优先；同标题、同作者的另一 arXiv 存档或扩展稿不会仅因相似标题被并入。没有身份锚点时，作者佐证的精确标题只有一个候选才可匹配；多个候选会报 `Ambiguous paper identity`，整个采集文件回滚并记录失败运行。应补有证据的标识或先核查版本关系，不删除重复候选或任取第一个。一个新来源明确同时携带两个既有稳定标识时，仍可按既有规则归并并保留投稿、标签和合并日志；未归并的同标题关系继续保留为待核查候选。这些规则不证明全库去重完成。 + ## 5. 投稿历史关系审计 ```sh diff --git a/paper_atlas/db/store.py b/paper_atlas/db/store.py index 7d57533..b40698f 100644 --- a/paper_atlas/db/store.py +++ b/paper_atlas/db/store.py @@ -95,13 +95,19 @@ class Store: matches.add(old[0]) key = normalized(title) title_matches = c.execute('SELECT id,authors_json,abstract FROM papers WHERE title_key=?',(key,)).fetchall() - authors_set = author_keys(authors) - for paper,old_authors,old_abstract in title_matches: - overlap = authors_set & author_keys(json.loads(old_authors)) - # Cross-platform title matching needs independent author corroboration. - same_source = c.execute('SELECT 1 FROM source_records WHERE paper_id=? AND source_id=? LIMIT 1',(paper,source_id)).fetchone() - if not same_source and len(key)>=25 and (len(overlap)>=2 or (overlap and abstract and abstract.strip()==(old_abstract or '').strip())): - matches.add(paper) + # An existing identifier/source record anchors the entity. Title + # similarity must not pull in another deposit or an extended version. + if not matches: + title_only_matches = set() + authors_set = author_keys(authors) + for paper,old_authors,old_abstract in title_matches: + overlap = authors_set & author_keys(json.loads(old_authors)) + same_source = c.execute('SELECT 1 FROM source_records WHERE paper_id=? AND source_id=? LIMIT 1',(paper,source_id)).fetchone() + if not same_source and len(key)>=25 and (len(overlap)>=2 or (overlap and abstract and abstract.strip()==(old_abstract or '').strip())): + title_only_matches.add(paper) + if len(title_only_matches)>1: + raise ValueError('Ambiguous paper identity: multiple author-corroborated title matches; provide verified identifiers') + matches = title_only_matches if matches: paper = min(matches) for other in sorted(matches-{paper}): diff --git a/tests/test_index_core.py b/tests/test_index_core.py index 0e18a06..dc9e2cf 100644 --- a/tests/test_index_core.py +++ b/tests/test_index_core.py @@ -53,6 +53,50 @@ class StoreTests(unittest.TestCase): self.assertNotEqual(other,paper) self.assertEqual(self.store.conn.execute('SELECT count(*) FROM match_candidates').fetchone()[0],1) + def test_identifier_anchor_does_not_absorb_a_same_title_arxiv_version(self): + conference,first=self.record('arxiv','1408.1485') + extended,second=self.record('arxiv','cs/0307069') + self.store.decision(extended,second,'ICLR',2025,'main','separate','https://example.org/reject','rejected','Reject') + resolved,_=self.record('dblp','conf/uai/example',{'arxiv':'1408.1485'}) + self.assertEqual(resolved,conference);self.assertNotEqual(conference,extended) + c=self.store.conn + self.assertEqual(c.execute('SELECT count(*) FROM papers').fetchone()[0],2) + self.assertEqual(c.execute('SELECT paper_id FROM submissions').fetchone()[0],extended) + self.assertEqual(c.execute('SELECT count(*) FROM paper_merges').fetchone()[0],0) + self.assertEqual(c.execute('SELECT count(*) FROM match_candidates').fetchone()[0],1) + # A refresh of the anchored original source is equally unambiguous. + self.assertEqual(self.record('arxiv','1408.1485')[0],conference) + self.assertEqual(c.execute('SELECT count(*) FROM papers').fetchone()[0],2) + + def test_ambiguous_title_aborts_whole_file_without_merging_existing_entities(self): + self.record('arxiv','1408.1485');self.record('arxiv','cs/0307069') + self.store.conn.commit() + rows=[{'source':'dblp','external_id':'conf/uai/new','url':'https://example.org/new', + 'title':'An unrelated earlier record in this file','abstract':None,'authors':[], + 'venue':'UAI','year':2001,'track':'main','status':'accepted'}, + {'source':'dblp','external_id':'conf/uai/ambiguous','url':'https://example.org/ambiguous', + 'title':'A sufficiently long example research paper title','abstract':None, + 'authors':['Jane Doe','John Smith'],'identifiers':{'doi':'10.0000/not-yet-known'}, + 'venue':'UAI','year':2001,'track':'main','status':'accepted'}] + path=Path(self.tmp.name)/'ambiguous.jsonl' + path.write_text(''.join(json.dumps(row)+'\n' for row in rows)) + path.with_suffix('.manifest.json').write_text(json.dumps({'source':'dblp','list_url':'https://example.org/list','count':2})) + with self.assertRaisesRegex(ValueError,'Ambiguous paper identity'):ingest_file(self.db,path) + c=self.store.conn + for table,count in [('papers',2),('source_records',2),('submissions',0),('decisions',0),('paper_merges',0)]: + self.assertEqual(c.execute('SELECT count(*) FROM '+table).fetchone()[0],count,table) + self.assertEqual(c.execute('SELECT status FROM ingest_runs').fetchone()[0],'failed') + + def test_explicit_bridge_merges_only_its_identifier_matches(self): + first,_=self.record('arxiv','2601.00001',authors=['First Author']) + second,_=self.record('arxiv','2601.00002',{'doi':'10.0000/bridge'},authors=['Second Author']) + unrelated,_=self.record('arxiv','2601.00003') + resolved,_=self.record('official','bridge',{'arxiv':'2601.00001','doi':'10.0000/bridge'}) + self.assertEqual(resolved,min(first,second)) + self.assertEqual(self.store.conn.execute('SELECT count(*) FROM papers').fetchone()[0],2) + self.assertIsNotNone(self.store.conn.execute('SELECT 1 FROM papers WHERE id=?',(unrelated,)).fetchone()) + self.assertEqual(self.store.conn.execute('SELECT count(*) FROM paper_merges').fetchone()[0],1) + def test_manifest_count_mismatch_rolls_back_records(self): path=Path(self.tmp.name)/'source.jsonl' path.write_text(json.dumps({'source':'pmlr','external_id':'v1/paper','url':'https://example.org/paper','title':'Paper','abstract':'Abstract','authors':[],'venue':'ICML','year':2023,'track':'main','status':'accepted'})+'\n') diff --git a/tests/test_postgres_ingestion.py b/tests/test_postgres_ingestion.py index c05bb0b..52334ad 100644 --- a/tests/test_postgres_ingestion.py +++ b/tests/test_postgres_ingestion.py @@ -110,6 +110,34 @@ class PostgreSQLIngestionTests(unittest.TestCase): finally: reader.close() + def test_identity_anchor_keeps_other_arxiv_version_and_ambiguous_file_rolls_back(self): + title='An identical long research title shared by conference and extended versions' + authors=['Jane Doe','John Smith'] + store=self.store() + try: + first,_=store.record('arxiv','1408.1485','https://arxiv.org/abs/1408.1485',title,'Conference abstract',authors) + second,_=store.record('arxiv','cs/0307069','https://arxiv.org/abs/cs/0307069',title,'Extended abstract',authors) + resolved,_=store.record('dblp','known','https://example.org/known',title,None,authors,{'arxiv':'1408.1485'}) + self.assertEqual(resolved,first);self.assertNotEqual(first,second) + store.conn.commit() + self.assertEqual(store.conn.execute('SELECT count(*) FROM papers').fetchone()[0],3) + self.assertEqual(store.conn.execute('SELECT count(*) FROM paper_merges').fetchone()[0],1) # migration fixture only + self.assertEqual(store.conn.execute('SELECT count(*) FROM match_candidates').fetchone()[0],1) + finally:store.conn.close() + rows=[self.row('official','earlier',status='accepted',venue='UAI',year=2001), + dict(self.row('official','ambiguous',status='accepted',venue='UAI',year=2001), + title=title,abstract=None,authors=authors,identifiers={})] + with self.assertRaisesRegex(ValueError,'Ambiguous paper identity'):self.ingest(self.file('ambiguous-identity',rows)) + store=self.store() + try: + self.assertEqual(store.conn.execute('SELECT count(*) FROM papers').fetchone()[0],3) + self.assertEqual(store.conn.execute('SELECT count(*) FROM source_records').fetchone()[0],4) + self.assertEqual(store.conn.execute('SELECT count(*) FROM submissions').fetchone()[0],1) + self.assertEqual(store.conn.execute('SELECT count(*) FROM paper_merges').fetchone()[0],1) + self.assertEqual(store.conn.execute('SELECT status FROM ingest_runs ORDER BY id DESC LIMIT 1').fetchone()[0],'failed') + self.assertEqual(store.conn.execute("SELECT count(*) FROM source_records WHERE external_id IN (?,?)",('earlier','ambiguous')).fetchone()[0],0) + finally:store.conn.close() + def test_file_import_is_idempotent_and_refreshes_only_own_tags(self): row = self.row(); row['tags'] = [{'namespace': 'openreview_snapshot:area', 'value': 'theory'}] path = self.file('reject', [row]); self.ingest(path) From ee140cfd40f298123cd0bee679b350c419c080de Mon Sep 17 00:00:00 2001 From: Codex Bot Date: Sun, 8 Nov 2026 17:31:12 +0800 Subject: [PATCH 2/3] Preserve source coverage when importing incremental metadata batches --- docs/operations.md | 33 ++++++++ paper_atlas/ingestion/records.py | 73 +++++++++++++++++- tests/test_incremental_coverage.py | 116 +++++++++++++++++++++++++++++ tests/test_postgres_ingestion.py | 48 ++++++++++++ 4 files changed, 268 insertions(+), 2 deletions(-) create mode 100644 tests/test_incremental_coverage.py diff --git a/docs/operations.md b/docs/operations.md index e4d8d64..7c302e0 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -120,3 +120,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 52334ad..9c516fd 100644 --- a/tests/test_postgres_ingestion.py +++ b/tests/test_postgres_ingestion.py @@ -87,6 +87,54 @@ 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_native_writes_keep_nul_historical_ids_and_later_acceptance(self): store = self.store() try: From 71d3cf0590c0c6ea7a63a587e83a5a436fe59641 Mon Sep 17 00:00:00 2001 From: Codex Bot Date: Sun, 8 Nov 2026 17:33:59 +0800 Subject: [PATCH 3/3] Verify incremental coverage rejects a concurrent parent update --- tests/test_postgres_ingestion.py | 28 ++++++++++++++++++++++++++++ 1 file changed, 28 insertions(+) diff --git a/tests/test_postgres_ingestion.py b/tests/test_postgres_ingestion.py index 9c516fd..1998e1d 100644 --- a/tests/test_postgres_ingestion.py +++ b/tests/test_postgres_ingestion.py @@ -135,6 +135,34 @@ class PostgreSQLIngestionTests(unittest.TestCase): 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: