From a3b677bdf29822d6bee81769df26676f963cbf49 Mon Sep 17 00:00:00 2001 From: Codex Bot Date: Sat, 7 Nov 2026 02:09:19 +0800 Subject: [PATCH] Add verified PostgreSQL migration and shared research queries --- .gitea/workflows/check.yml | 8 +- README.md | 13 +- docs/operations.md | 26 +++ docs/usage.md | 16 ++ paper_atlas/cli.py | 1 + paper_atlas/db/backend.py | 8 + paper_atlas/db/store.py | 2 + paper_atlas/ingestion/arxiv_tags.py | 2 + paper_atlas/ingestion/maintenance.py | 2 + paper_atlas/ingestion/pmlr_tracks.py | 3 + paper_atlas/postgres/__init__.py | 1 + paper_atlas/postgres/client.py | 83 ++++++++++ paper_atlas/postgres/migration.py | 232 +++++++++++++++++++++++++++ paper_atlas/postgres/schema.py | 120 ++++++++++++++ paper_atlas/querying.py | 30 +++- pyproject.toml | 3 + tests/test_postgres.py | 212 ++++++++++++++++++++++++ 17 files changed, 748 insertions(+), 14 deletions(-) create mode 100644 paper_atlas/db/backend.py create mode 100644 paper_atlas/postgres/__init__.py create mode 100644 paper_atlas/postgres/client.py create mode 100644 paper_atlas/postgres/migration.py create mode 100644 paper_atlas/postgres/schema.py create mode 100644 tests/test_postgres.py diff --git a/.gitea/workflows/check.yml b/.gitea/workflows/check.yml index 7fb5f67..0eeabc1 100644 --- a/.gitea/workflows/check.yml +++ b/.gitea/workflows/check.yml @@ -30,11 +30,16 @@ jobs: if len(candidates)!=1: raise SystemExit('Unexpected repository archive layout') src = candidates[0].parent - subprocess.run(['python3','-B','-m','unittest','discover','-s','tests','-t','.','-v'],cwd=src,check=True) # Build and installation checks run only on the approved remote runner. venv = root / 'venv' subprocess.run(['python3','-B','-m','venv','--system-site-packages',str(venv)],check=True) python = str(venv / 'bin/python') + subprocess.run([python,'-B','-m','pip','install','--only-binary=:all:','--no-compile','psycopg[binary]==3.3.8'],check=True) + testenv = os.environ.copy() + service_file = pathlib.Path('/etc/paper-atlas/pg_service.conf') + if service_file.is_file(): + testenv.update(PGSERVICEFILE=str(service_file),PAPER_ATLAS_TEST_PG_SERVICE='foundry-research') + subprocess.run([python,'-B','-m','unittest','discover','-s','tests','-t','.','-v'],cwd=src,env=testenv,check=True) wheels = root / 'wheels' subprocess.run([python,'-B','-m','pip','wheel','--no-deps','--wheel-dir',str(wheels),str(src)],check=True) wheel, = wheels.glob('paper_atlas-*.whl') @@ -43,6 +48,7 @@ jobs: outside.mkdir() subprocess.run([str(venv/'bin/paper-atlas'),'--help'],cwd=outside,check=True) subprocess.run([python,'-I','-B','-m','paper_atlas','collect','cvf','--help'],cwd=outside,check=True) + subprocess.run([python,'-I','-B','-m','paper_atlas','maintain','postgres','--help'],cwd=outside,check=True) smoke = "from paper_atlas.db.store import Store; from paper_atlas.config import DEFAULT_REGISTRY; import json; assert {'COLM','MM','MLSys'} <= {v['acronym'] for v in json.loads(DEFAULT_REGISTRY.read_text())['venues']}; s=Store('installed.sqlite'); s.conn.close()" subprocess.run([python,'-I','-B','-c',smoke],cwd=outside,check=True) subprocess.run([python,'-I','-B','-m','paper_atlas','query','--db','installed.sqlite','--stats'],cwd=outside,check=True) diff --git a/README.md b/README.md index e06dfcf..43ac6bc 100644 --- a/README.md +++ b/README.md @@ -2,7 +2,7 @@ **长期维护、按需运行的论文元数据与公开投稿历史工具。** -Paper Atlas 将 arXiv、AI 会议论文集及公开投稿决定整理到可追溯的 SQLite 数据库中，用于查询和后续研究分析。它按批次运行，不需要常驻 Web 服务；采集标题、摘要、标识、标签及决定证据，不下载论文全文。 +Paper Atlas 将 arXiv、AI 会议论文集及公开投稿决定整理到可追溯的数据库中，用于查询和后续研究分析。生产数据使用 PostgreSQL，支持从既有 SQLite 索引完整迁移；SQLite 保留离线核验和小型样本用途。它按批次运行，不需要常驻 Web 服务；采集标题、摘要、标识、标签及决定证据，不下载论文全文。 项目仍在建设中。候选范围包含 43 个会议（含 ACM MM、MLSys），尚未完成全部历史数据和最终去重复核。实际进度以数据库、来源清单和覆盖报告为准，而不是仓库文件数量。 @@ -14,6 +14,7 @@ paper_atlas/ collectors/ 各来源采集与目录盘点 ingestion/ 导入、身份回填与单写入协调 db/ 数据模型、身份解析与 SQL schema + postgres/ PostgreSQL 迁移、原生全文索引和只读连接 audits/ 来源核验和覆盖报告 io/ 网络、快照下载、HTML 与 Git 元数据读取 resources/ 候选会议目录 @@ -30,7 +31,7 @@ pyproject.toml 包元数据、依赖和命令注册 ```sh python3 -m venv .venv -.venv/bin/python -m pip install . +.venv/bin/python -m pip install '.[postgres]' source .venv/bin/activate paper-atlas --help ``` @@ -41,13 +42,13 @@ paper-atlas --help ```sh # 查看当前入库量；不同平台记录数不等于独立论文数 -paper-atlas query --db /data/paper-index/papers.sqlite --stats +PGSERVICEFILE=/etc/paper-atlas/pg_service.conf paper-atlas query --pg-service foundry-research --stats # 查询标题与摘要 -paper-atlas query --db /data/paper-index/papers.sqlite --query '"language model"' +PGSERVICEFILE=/etc/paper-atlas/pg_service.conf paper-atlas query --pg-service foundry-research --query '"language model"' # 查看一篇论文的来源、标签和所有已知投稿历史 -paper-atlas query --db /data/paper-index/papers.sqlite --paper 123 +PGSERVICEFILE=/etc/paper-atlas/pg_service.conf paper-atlas query --pg-service foundry-research --paper 123 # 列出全部目标会议，包括尚未入库的项目 paper-atlas coverage --db /data/paper-index/papers.sqlite \ @@ -60,6 +61,8 @@ paper-atlas collect colm --cache /data/paper-index/raw/http-cache \ `paper-atlas collect --help` 查看来源；`paper-atlas import --help` 查看导入方式。完整批次步骤见[使用指南](docs/usage.md)。 +共享实例名为 `foundry-postgres`，数据位于 `foundry_data` 库的 `research` schema；其他项目可按权限复用同一组表。库名通过 libpq service 配置、schema 通过 search_path 或 `--pg-schema` 指定，不写死在查询中。PostgreSQL 凭据放在服务器权限为 `0600` 的 libpq service 文件中；命令行只使用服务名称。`paper-atlas maintain postgres --help` 提供迁移入口：先完整核验 21 张业务表，再发布目标 schema。SQLite FTS5 内部表由 PostgreSQL GIN 索引替代。切换后 SQLite 为备份，不能继续用旧导入器写入它；后续生产导入须适配 PostgreSQL，不能把 SQLite 更新当作生产更新。 + ## 数据如何组织 - **论文和投稿分开。** 一篇论文可以多次投稿；某次被拒、后来在其他会议录用，两段历史都保留。 diff --git a/docs/operations.md b/docs/operations.md index 91e73fe..6ac8f5a 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -1,5 +1,31 @@ # 运行维护与数据核验 +## PostgreSQL 迁移 + +共享实例的容器名为 `foundry-postgres`，底层镜像为 `postgres:17-alpine`；Compose 服务名仍为 `postgres`。数据库和 schema 是可配置的逻辑边界，表按研究数据域组织，不按调用项目复制。其他项目可获授 `research` 的 USAGE 及必要表权限，复用相同论文 ID 和表。未来换库只更新 libpq service；换 schema 可更新角色 search_path 或查询的 `--pg-schema`。 + +复用已确认的现有实例，在共享 `foundry_data` 数据库按数据域使用 `research` schema 和专用角色，不修改其他业务库。安装、测试和构建在已批准的 `ces-control:host` 执行环境运行。libpq service 文件由服务器提供，权限必须为 `0600`，不得放进仓库或带密码的命令行参数。 + +冻结 SQLite 写入，确认维护进程实际结束、WAL 已清空；采集到独立文件不修改索引。迁移持有源维护锁和目标 advisory lock，还须确保没有绕过协调器的旧版手工写入。 + +```sh +export PGSERVICEFILE=/etc/paper-atlas/pg_service.conf +paper-atlas maintain postgres \ + --db /data/paper-index/papers.sqlite --pg-service foundry-research \ + --storage-path /var/lib/docker/volumes/endoleap-dev_postgres/_data \ + --min-free-gib 8 --out /data/paper-index/collected/postgres-migration.json +``` + +全部 21 张业务表保留原 ID、原始 JSON 文本、NULL、来源、投稿、决定、标签出处和合并记录。`sqlite_sequence` 用于重设 PostgreSQL identity 序列，并考虑已合并的历史 ID，避免复用；SQLite FTS5 shadow 表根据完整标题/摘要重建为原生 GIN 索引，不逐字复制引擎文件。 + +先 COPY 到独立暂存 schema，每张表提交和记录哈希。中断后以同一命令、同一报告路径恢复，必须仍是**完全相同的未变更快照**。不同输入、已发布目标、未清空 WAL、新的未知业务表都会拒绝覆盖；恢复只会清空本次运行尚未完成且尚未发布的暂存表。保留错误现场，不盲目从头复制。 + +发布前逐表按主键完整回读，比较数量及逐行 SHA-256；建立并验证所有主键、唯一约束、CHECK 和外键，重设序列、建立 GIN 并 ANALYZE。全部通过后原子重命名为 `research` schema，报告才为 `complete`。`copying`、`prepared`、部分表通过或 CI 通过均不代表生产迁移完成，也不替代原任务最后的实体去重和会议覆盖审计。 + +发布后核对 PostgreSQL 的统计、全文检索、论文详情，以及 reject 后另会 accept 的历史。确认后保留 SQLite 只读备份，并在旁保存 `.postgresql-cutover.json`，记录报告路径、源哈希、目标 service/schema；新版本旧写入入口见此标记会拒绝写入。标记不含密码。后续导入适配属于下一阶段，不可继续更新 SQLite 后声称 PostgreSQL 已同步。 + +远程集成测试通过 `PAPER_ATLAS_TEST_PG_SERVICE` 显式指定，使用随机前缀 schema 并仅清理自己的 fixture；没有该配置时会跳过，应另外报告“未运行”，不能用 SQLite 测试代替 PostgreSQL 验证。 + 代码以固定 commit 部署，已经启动的任务继续使用其原版本。目录重构不要求迁移数据库，也不更改已有 JSONL 和 manifest 格式。 ## arXiv 原始编号与类别证据 diff --git a/docs/usage.md b/docs/usage.md index f0a84cf..12503f6 100644 --- a/docs/usage.md +++ b/docs/usage.md @@ -2,6 +2,22 @@ 先在已确认的远程环境安装项目，使用 `paper-atlas`。从源码运行时，将命令前缀替换为 `python3 -B -m paper_atlas`，其余参数相同。数据目录位于仓库外。 +## PostgreSQL 查询 + +在远程环境安装 `.[postgres]`，通过权限为 `0600` 的 libpq service 文件连接，不把密码放入 URL、命令行或仓库。 + +```sh +export PGSERVICEFILE=/etc/paper-atlas/pg_service.conf +paper-atlas query --pg-service foundry-research --stats +paper-atlas query --pg-service foundry-research --query '"language model"' --tag arxiv:category=cs.CL +paper-atlas query --pg-service foundry-research --status rejected --outcome accepted --limit 20 +paper-atlas query --pg-service foundry-research --paper 123 +``` + +使用只读一致性事务，支持论文 ID、会议别名、年份、轨道、标签、最新决定和跨会议录用筛选。全文检索使用 PostgreSQL `websearch_to_tsquery('english', ...)`，支持引号短语、`OR` 和减号排除；与 SQLite FTS5 的语法及分词结果不完全相同。JSON 来源列保留原始文本，SQL 分析时可用 `metadata_json::jsonb`、`authors_json::jsonb`。 + +迁移与切换见[运行维护](operations.md#postgresql-迁移)。下述 SQLite 导入命令仅用于未切换的索引或独立样本；切换后的生产源拒绝旧写入。尚未适配的覆盖审计可只读检查冻结备份，不能当作 PostgreSQL 当前状态。 + ## 1. 下载固定快照 `paper-atlas download --help` 说明大小、校验和与续传参数。各来源的固定版本与具体命令见[数据源说明](data-sources.md)。下载成功只表示原始文件已核验。 diff --git a/paper_atlas/cli.py b/paper_atlas/cli.py index 4db7448..fe0ec53 100644 --- a/paper_atlas/cli.py +++ b/paper_atlas/cli.py @@ -29,6 +29,7 @@ COMMANDS = { ('maintain','backfill'): ('paper_atlas.ingestion.backfill', {}, '回填既有投稿的来源身份'), ('maintain','arxiv-tags'): ('paper_atlas.ingestion.arxiv_tags', {}, '恢复逐来源记录的 arXiv 类别证据'), ('maintain','pmlr-tracks'): ('paper_atlas.ingestion.pmlr_tracks', {}, '核对并迁移既有 PMLR 投稿轨道'), + ('maintain','postgres'): ('paper_atlas.postgres.migration', {}, '迁移 SQLite 完整业务数据并核验 PostgreSQL'), ('audit','iclr'): ('paper_atlas.audits.iclr', {}, '核对 ICLR 快照与同届官方程序'), ('audit','iclr-legacy'): ('paper_atlas.audits.iclr_legacy', {}, '独立核验 ICLR 2020 旧目录投稿身份'), ('audit','arxiv'): ('paper_atlas.audits.arxiv', {}, '核验 arXiv 原始编号、重复行与 ZIP 完整性'), diff --git a/paper_atlas/db/backend.py b/paper_atlas/db/backend.py new file mode 100644 index 0000000..3faf2ad --- /dev/null +++ b/paper_atlas/db/backend.py @@ -0,0 +1,8 @@ +"""Prevent legacy writers from diverging after an explicit PostgreSQL cutover.""" +from pathlib import Path + + +def require_sqlite_writable(path): + marker = Path(str(Path(path).resolve()) + '.postgresql-cutover.json') + if marker.exists(): + raise RuntimeError('This SQLite source was cut over to PostgreSQL; use the configured PostgreSQL database. Marker: ' + str(marker)) diff --git a/paper_atlas/db/store.py b/paper_atlas/db/store.py index 8cbf4a1..9495ecc 100644 --- a/paper_atlas/db/store.py +++ b/paper_atlas/db/store.py @@ -6,6 +6,7 @@ import sqlite3 import unicodedata from pathlib import Path from paper_atlas.db.submissions import attach_submission,migrate_sources,write_decision +from paper_atlas.db.backend import require_sqlite_writable def normalized(value): @@ -29,6 +30,7 @@ def author_keys(authors): class Store: def __init__(self,path): + require_sqlite_writable(path) self.conn = sqlite3.connect(path,timeout=60) self.conn.executescript(files('paper_atlas.db').joinpath('schema.sql').read_text()) self.conn.execute('PRAGMA synchronous=NORMAL') diff --git a/paper_atlas/ingestion/arxiv_tags.py b/paper_atlas/ingestion/arxiv_tags.py index 2e5cdc9..9c6b846 100644 --- a/paper_atlas/ingestion/arxiv_tags.py +++ b/paper_atlas/ingestion/arxiv_tags.py @@ -18,6 +18,8 @@ def save(path,state): def repair(db,out,resume=False,batch_size=1000): db=Path(db).resolve();out=Path(out).resolve() + from paper_atlas.db.backend import require_sqlite_writable + require_sqlite_writable(db) if not db.is_file():raise ValueError('Existing database required') if batch_size<1 or batch_size>10000:raise ValueError('Batch size must be between 1 and 10000') out.parent.mkdir(parents=True,exist_ok=True) diff --git a/paper_atlas/ingestion/maintenance.py b/paper_atlas/ingestion/maintenance.py index 829f5b8..0be9d2e 100644 --- a/paper_atlas/ingestion/maintenance.py +++ b/paper_atlas/ingestion/maintenance.py @@ -15,6 +15,7 @@ from pathlib import Path from paper_atlas.ingestion.records import ingest_file from paper_atlas.db.store import Store from paper_atlas.ingestion.backfill import backfill,queue_files +from paper_atlas.db.backend import require_sqlite_writable UNIT='paper-atlas-arxiv-import.service' @@ -25,6 +26,7 @@ def property_value(name): def merge_locked(db,archive,files,backfill_queue=None): + require_sqlite_writable(db) # Validate inputs before stopping the existing writer. for filename in files: path=Path(filename) diff --git a/paper_atlas/ingestion/pmlr_tracks.py b/paper_atlas/ingestion/pmlr_tracks.py index a528dc0..dbb2134 100644 --- a/paper_atlas/ingestion/pmlr_tracks.py +++ b/paper_atlas/ingestion/pmlr_tracks.py @@ -123,6 +123,9 @@ def plan(c, scopes): def reclassify(db, inventories, out, apply=False): db, out = Path(db).resolve(), Path(out).resolve() + if apply: + from paper_atlas.db.backend import require_sqlite_writable + require_sqlite_writable(db) if not db.is_file() or out == db: raise ValueError('Existing database and distinct report path required') scopes = load_inventories(inventories) diff --git a/paper_atlas/postgres/__init__.py b/paper_atlas/postgres/__init__.py new file mode 100644 index 0000000..2e81e98 --- /dev/null +++ b/paper_atlas/postgres/__init__.py @@ -0,0 +1 @@ +"""PostgreSQL migration and read access for the paper index.""" diff --git a/paper_atlas/postgres/client.py b/paper_atlas/postgres/client.py new file mode 100644 index 0000000..076c6c5 --- /dev/null +++ b/paper_atlas/postgres/client.py @@ -0,0 +1,83 @@ +"""Libpq service connections; credentials never belong in CLI arguments.""" +from collections.abc import Mapping +import re + +SEARCH_VECTOR = "to_tsvector('english'::regconfig,coalesce(title,'') || ' ' || coalesce(abstract,''))" + + +def driver(): + try: + import psycopg + except ImportError as error: + raise RuntimeError('Install paper-atlas[postgres] on the approved remote runner') from error + return psycopg + + +def connect(service, *, readonly=False): + if not re.fullmatch(r'[A-Za-z0-9_-]+', service): + raise ValueError('Use a libpq service name, never a password or connection URL') + connection = driver().connect(service=service, connect_timeout=15, autocommit=True, + application_name='paper-atlas') + connection.execute("SET timezone='UTC'") + connection.execute("SET lock_timeout='5s'") + connection.execute("SET work_mem='2MB'") + connection.execute("SET maintenance_work_mem='16MB'") + connection.execute('SET max_parallel_workers_per_gather=0') + connection.execute('SET max_parallel_maintenance_workers=0') + if readonly: + connection.execute('BEGIN ISOLATION LEVEL REPEATABLE READ READ ONLY') + return connection + + +class Row(Mapping): + """The small mapping/positional interface used by existing query output.""" + def __init__(self, columns, values): + self.columns, self.values = columns, values + + def __getitem__(self, key): + return self.values[key if isinstance(key, int) else self.columns.index(key)] + + def __iter__(self): + return iter(self.columns) + + def __len__(self): + return len(self.columns) + + +def row_factory(cursor): + columns = [column.name for column in cursor.description] if cursor.description else [] + return lambda values: Row(columns, values) + + +class ReadConnection: + postgres = True + + def __init__(self, service, schema=None): + if schema is not None and not re.fullmatch(r'[a-z][a-z0-9_]{0,35}', schema): + raise ValueError('Invalid schema name') + self.connection = connect(service, readonly=True) + from paper_atlas.postgres.schema import quote + if schema is not None: + self.connection.execute('SET LOCAL search_path TO ' + quote(schema) + ',pg_catalog') + self.connection.row_factory = row_factory + if not self.has_table('papers'): + self.close() + raise RuntimeError('PostgreSQL publication is not ready in the configured search_path') + + def execute(self, query, params=()): + # All caller SQL is static; user values are always bound parameters. + if 'sqlite_master' in query: + match = re.search(r"name='([a-z_]+)'", query) + if not match or match[1] not in ('decision_sources', 'submission_sources', 'venue_aliases'): + raise ValueError('Unsupported legacy table lookup') + return self.connection.execute('''SELECT 1 FROM information_schema.tables + WHERE table_schema=current_schema() AND table_name=%s''', + (match[1],)) + return self.connection.execute(query.replace('?', '%s'), params) + + def has_table(self, name): + return bool(self.connection.execute('''SELECT 1 FROM information_schema.tables + WHERE table_schema=current_schema() AND table_name=%s''', (name,)).fetchone()) + + def close(self): + self.connection.close() diff --git a/paper_atlas/postgres/migration.py b/paper_atlas/postgres/migration.py new file mode 100644 index 0000000..be97f52 --- /dev/null +++ b/paper_atlas/postgres/migration.py @@ -0,0 +1,232 @@ +"""Resumable, unpublished COPY followed by complete ordered content verification.""" +import argparse +from contextlib import closing +import fcntl +import hashlib +import json +from pathlib import Path +import re +import shutil +import sqlite3 +import uuid +import zlib + +from paper_atlas.db.submissions import timestamp +from paper_atlas.postgres.client import connect, SEARCH_VECTOR +from paper_atlas.postgres.schema import TABLES, inspect, fingerprint, quote, select_sql, table_sql + + +def digest_file(path): + digest = hashlib.sha256() + with path.open('rb') as stream: + while block := stream.read(8 * 1024 ** 2): + digest.update(block) + return digest.hexdigest() + + +def update_hash(digest, row): + # Raw JSON columns remain text: whitespace, escapes and source evidence survive. + data = json.dumps(tuple(row), ensure_ascii=False, separators=(',', ':')).encode() + digest.update(len(data).to_bytes(8, 'big')); digest.update(data) + + +def save(path, result): + path.parent.mkdir(parents=True, exist_ok=True) + part = path.with_suffix(path.suffix + '.part') + part.write_text(json.dumps(result, ensure_ascii=False, indent=2) + '\n') + part.replace(path) + + +def free_space(path, minimum): + if path and shutil.disk_usage(path).free < minimum * 1024 ** 3: + raise RuntimeError('Migration storage headroom fell below the configured minimum') + + +def qualify(schema, table): + return quote(schema) + '.' + quote(table) + + +def add_constraints(pg, stage, plan): + def constraint(table, name, clause): + exists = pg.execute('''SELECT 1 FROM pg_constraint co JOIN pg_class cl ON cl.oid=co.conrelid + JOIN pg_namespace ns ON ns.oid=cl.relnamespace WHERE ns.nspname=%s AND cl.relname=%s AND co.conname=%s''', + (stage, table, name)).fetchone() + if not exists: + pg.execute('ALTER TABLE ' + qualify(stage, table) + ' ADD CONSTRAINT ' + quote(name) + ' ' + clause) + + for table, spec in plan.items(): + constraint(table, table + '_pkey', 'PRIMARY KEY (' + ','.join(map(quote, spec['pk'])) + ')') + for number, index in enumerate(spec['indexes']): + if index['origin'] == 'u': + constraint(table, table + '_unique_' + str(number), 'UNIQUE (' + ','.join(quote(n) for n, desc in index['terms']) + ')') + else: + terms = ','.join(quote(n) + (' DESC' if desc else '') for n, desc in index['terms']) + pg.execute('CREATE ' + ('UNIQUE ' if index['unique'] else '') + 'INDEX IF NOT EXISTS ' + + quote(index['name']) + ' ON ' + qualify(stage, table) + '(' + terms + ')') + for table, spec in plan.items(): + for number, fk in enumerate(spec['foreign_keys']): + clause = ('FOREIGN KEY (' + ','.join(map(quote, fk['columns'])) + ') REFERENCES ' + + qualify(stage, fk['table']) + '(' + ','.join(map(quote, fk['references'])) + ') NOT VALID') + constraint(table, table + '_fk_' + str(number), clause) + for number, check in enumerate(spec['checks']): + constraint(table, table + '_check_' + str(number), 'CHECK (' + check + ') NOT VALID') + constraints = pg.execute('''SELECT cl.relname,co.conname FROM pg_constraint co + JOIN pg_class cl ON cl.oid=co.conrelid JOIN pg_namespace ns ON ns.oid=cl.relnamespace + WHERE ns.nspname=%s AND NOT co.convalidated ORDER BY cl.relname,co.conname''', (stage,)).fetchall() + for table, name in constraints: + pg.execute('ALTER TABLE ' + qualify(stage, table) + ' VALIDATE CONSTRAINT ' + quote(name)) + + +def sequences(pg, source, stage, plan): + old = dict(source.execute('SELECT name,seq FROM sqlite_sequence')) + history = {'papers': ('paper_merges', 'former_paper_id'), + 'submissions': ('submission_merges', 'former_submission_id'), + 'decisions': ('decision_merges', 'former_decision_id')} + result = {} + for table, spec in plan.items(): + if spec['pk'] != ['id']: + continue + value = max(source.execute('SELECT coalesce(max(id),0) FROM ' + quote(table)).fetchone()[0], old.get(table, 0)) + if table in history: + audit, column = history[table] + value = max(value, source.execute('SELECT coalesce(max(' + quote(column) + '),0) FROM ' + quote(audit)).fetchone()[0]) + sequence = pg.execute('SELECT pg_get_serial_sequence(%s,%s)', (qualify(stage, table), 'id')).fetchone()[0] + pg.execute('SELECT setval(%s,%s,%s)', (sequence, max(value, 1), value > 0)) + result[table] = value + return result + + +def migrate(db, service, out, *, schema='research', storage_path=None, min_free_gib=8): + if not re.fullmatch(r'[a-z][a-z0-9_]{0,35}', schema): + raise ValueError('Use a short lowercase target schema name') + db, out = Path(db).resolve(), Path(out).resolve() + if not db.is_file() or out == db: + raise ValueError('Existing source and a distinct report are required') + if out.exists() and json.loads(out.read_text()).get('status') == 'complete': + raise ValueError('Completed report must be preserved') + free_space(storage_path, min_free_gib) + control = schema + '_migration' + report = {'status': 'preflight', 'source': str(db), 'service': service, 'schema': schema, + 'started_at': timestamp(), 'tables': {}, 'scope': 'All 21 business tables; FTS5 internals rebuilt as native PostgreSQL GIN.'} + with Path(str(db) + '.maintenance.lock').open('a') as lock: + fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB) + if Path(str(db) + '-wal').exists() and Path(str(db) + '-wal').stat().st_size: + raise ValueError('Freeze writers and checkpoint SQLite WAL before migrating') + with closing(sqlite3.connect(db.as_uri() + '?mode=ro', uri=True)) as source, closing(connect(service)) as pg: + source.execute('PRAGMA query_only=ON'); source.execute('PRAGMA cache_size=-16384'); source.execute('BEGIN') + plan = inspect(source) + stat = db.stat() + identity = {'file_sha256': digest_file(db), 'bytes': stat.st_size, 'mtime_ns': stat.st_mtime_ns, + 'schema_sha256': fingerprint(plan)} + report['source_identity'] = identity + advisory = zlib.crc32(schema.encode()) & 0x7fffffff + if not pg.execute('SELECT pg_try_advisory_lock(%s,%s)', (615173, advisory)).fetchone()[0]: + raise RuntimeError('Another migration holds the target lock') + if pg.execute('SELECT 1 FROM pg_namespace WHERE nspname=%s', (schema,)).fetchone(): + raise ValueError('Target schema already exists; published data will never be overwritten') + pg.execute('CREATE SCHEMA IF NOT EXISTS ' + quote(control)) + pg.execute('CREATE TABLE IF NOT EXISTS ' + qualify(control, 'run') + + '(id INTEGER PRIMARY KEY CHECK(id=1), source_json TEXT NOT NULL, stage TEXT NOT NULL, status TEXT NOT NULL)') + pg.execute('CREATE TABLE IF NOT EXISTS ' + qualify(control, 'tables') + + '(name TEXT PRIMARY KEY, rows BIGINT NOT NULL, sha256 TEXT NOT NULL, verified BOOLEAN NOT NULL DEFAULT false)') + run = pg.execute('SELECT source_json,stage,status FROM ' + qualify(control, 'run') + ' WHERE id=1').fetchone() + if run: + if json.loads(run[0]) != identity or run[2] == 'complete': + raise ValueError('Resume requires the exact same unpublished SQLite snapshot') + stage = run[1] + else: + stage = schema + '_stage_' + uuid.uuid4().hex[:12] + with pg.transaction(): + pg.execute('CREATE SCHEMA ' + quote(stage)) + for table, spec in plan.items(): + pg.execute(table_sql(stage, table, spec)) + pg.execute('INSERT INTO ' + qualify(control, 'run') + ' VALUES(1,%s,%s,%s)', + (json.dumps(identity, sort_keys=True), stage, 'copying')) + report.update(status='copying', stage=stage); save(out, report) + try: + for table, spec in plan.items(): + free_space(storage_path, min_free_gib) + previous = pg.execute('SELECT rows,sha256 FROM ' + qualify(control, 'tables') + ' WHERE name=%s', (table,)).fetchone() + if previous: + rows, digest = previous + else: + digest = hashlib.sha256(); rows = 0 + with pg.transaction(): + # Only this run's unpublished, incomplete table can be reset. + pg.execute('TRUNCATE ' + qualify(stage, table)) + columns = ','.join(quote(c['name']) for c in spec['columns']) + with pg.cursor().copy('COPY ' + qualify(stage, table) + '(' + columns + ') FROM STDIN') as copy: + for row in source.execute(select_sql(None, table, spec)): + copy.write_row(row); update_hash(digest, row); rows += 1 + if rows % 10000 == 0: + free_space(storage_path, min_free_gib) + if rows % 100000 == 0: + print(json.dumps({'stage': 'copying', 'table': table, 'rows': rows}), flush=True) + digest = digest.hexdigest() + pg.execute('INSERT INTO ' + qualify(control, 'tables') + '(name,rows,sha256) VALUES(%s,%s,%s)', (table, rows, digest)) + report['tables'][table] = {'source_rows': rows, 'source_sha256': digest} + save(out, report) + print(json.dumps({'stage': 'copied', 'table': table, 'rows': rows}), flush=True) + report['status'] = 'validating_constraints'; save(out, report) + add_constraints(pg, stage, plan) + report['sequences'] = sequences(pg, source, stage, plan) + report['status'] = 'verifying_contents'; save(out, report) + for table, spec in plan.items(): + digest = hashlib.sha256(); rows = 0 + with pg.transaction(), pg.cursor(name='verify_' + table) as cursor: + cursor.itersize = 512 + cursor.execute(select_sql(stage, table, spec)) + for row in cursor: + update_hash(digest, row); rows += 1 + entry = report['tables'][table] + entry.update(target_rows=rows, target_sha256=digest.hexdigest()) + if rows != entry['source_rows'] or digest.hexdigest() != entry['source_sha256']: + raise ValueError('Complete content verification failed for ' + table) + pg.execute('UPDATE ' + qualify(control, 'tables') + ' SET verified=true WHERE name=%s', (table,)) + save(out, report) + print(json.dumps({'stage': 'verified', 'table': table, 'rows': rows}), flush=True) + report['status'] = 'building_search_index'; save(out, report) + free_space(storage_path, min_free_gib) + pg.execute('CREATE INDEX IF NOT EXISTS papers_search_gin ON ' + qualify(stage, 'papers') + ' USING gin (' + SEARCH_VECTOR + ')') + search = pg.execute('''SELECT am.amname,i.indisvalid,i.indisready,t.relname + FROM pg_index i JOIN pg_class ix ON ix.oid=i.indexrelid + JOIN pg_class t ON t.oid=i.indrelid JOIN pg_namespace ns ON ns.oid=ix.relnamespace + JOIN pg_am am ON am.oid=ix.relam WHERE ns.nspname=%s AND ix.relname='papers_search_gin' ''', (stage,)).fetchone() + if search != ('gin', True, True, 'papers'): + raise ValueError('Native full-text index is not valid and ready') + for table in TABLES: + pg.execute('ANALYZE ' + qualify(stage, table)) + invalid = pg.execute('''SELECT count(*) FROM pg_constraint co JOIN pg_class cl ON cl.oid=co.conrelid + JOIN pg_namespace ns ON ns.oid=cl.relnamespace WHERE ns.nspname=%s AND NOT co.convalidated''', (stage,)).fetchone()[0] + if invalid: + raise ValueError('Unvalidated constraints remain') + if (db.stat().st_size, db.stat().st_mtime_ns) != (identity['bytes'], identity['mtime_ns']): + raise ValueError('Source changed during migration') + if Path(str(db) + '-wal').exists() and Path(str(db) + '-wal').stat().st_size: + raise ValueError('A source writer committed during migration') + report.update(status='prepared', foreign_keys='all_validated', checks='all_validated', native_search_index=True) + save(out, report) + with pg.transaction(): + pg.execute('ALTER SCHEMA ' + quote(stage) + ' RENAME TO ' + quote(schema)) + pg.execute('UPDATE ' + qualify(control, 'run') + " SET status='complete' WHERE id=1") + report.update(status='complete', finished_at=timestamp()) + save(out, report) + return report + except BaseException as error: + report.update(status='error', error_type=type(error).__name__, error=str(error), failed_at=timestamp()) + save(out, report) + raise + + +def main(argv=None, *, prog=None): + parser = argparse.ArgumentParser(prog=prog) + parser.add_argument('--db', required=True, help='Frozen SQLite source') + parser.add_argument('--pg-service', required=True, help='Protected libpq service name') + parser.add_argument('--schema', default='research') + parser.add_argument('--out', required=True) + parser.add_argument('--storage-path', help='Target filesystem path for free-space checks') + parser.add_argument('--min-free-gib', type=float, default=8) + args = parser.parse_args(argv) + result = migrate(args.db, args.pg_service, args.out, schema=args.schema, + storage_path=args.storage_path, min_free_gib=args.min_free_gib) + print(json.dumps(result, ensure_ascii=False), flush=True) diff --git a/paper_atlas/postgres/schema.py b/paper_atlas/postgres/schema.py new file mode 100644 index 0000000..90a97de --- /dev/null +++ b/paper_atlas/postgres/schema.py @@ -0,0 +1,120 @@ +"""Inspect the complete business schema; rebuild SQLite engine internals natively.""" +import hashlib +import json +import re + +TABLES = ('sources', 'papers', 'identifiers', 'source_records', 'venues', 'editions', + 'venue_aliases', 'edition_sources', 'submissions', 'decisions', + 'submission_sources', 'submission_identifiers', 'decision_sources', + 'submission_merges', 'decision_merges', 'schema_migrations', 'tags', + 'paper_tags', 'match_candidates', 'paper_merges', 'ingest_runs') +CHECKS = { + 'venues': ["venue_type IN ('conference','journal','review_platform')"], + 'edition_sources': ["status IN ('pending','running','complete','partial','blocked','unavailable','not_yet_published')"], + 'decisions': ["status IN ('accepted','rejected','desk_rejected','withdrawn','under_review','unknown')"], + 'match_candidates': ["status IN ('pending','merged','distinct')"], +} +TYPES = {'INTEGER': 'BIGINT', 'TEXT': 'TEXT COLLATE "C"', 'REAL': 'DOUBLE PRECISION'} + + +def checks_in(ddl): + result = [] + for match in re.finditer(r'\bCHECK\s*\(', ddl, re.I): + start = position = match.end() + depth, quoted = 1, False + while position < len(ddl) and depth: + char = ddl[position] + if char == "'": + if quoted and position + 1 < len(ddl) and ddl[position + 1] == "'": + position += 2; continue + quoted = not quoted + elif not quoted: + depth += (char == '(') - (char == ')') + position += 1 + if depth: + raise ValueError('Unclosed CHECK constraint') + result.append(ddl[start:position - 1]) + return result + + +def normalize_check(expression): + # Normalize SQL layout/keywords, retaining case and spaces inside literals. + tokens = re.findall(r"'(?:[^']|'')*'|[^']+", expression) + return ''.join(token if token.startswith("'") else re.sub(r'\s+', '', token).lower() for token in tokens) + + +def quote(name): + return '"' + name.replace('"', '""') + '"' + + +def inspect(connection): + business = {name for schema, name, kind, *_ in connection.execute('PRAGMA table_list') + if schema == 'main' and kind == 'table' and not name.startswith('sqlite_')} + if business != set(TABLES): + raise ValueError('Unexpected business tables: ' + repr(sorted(business ^ set(TABLES)))) + result = {} + for table in TABLES: + columns = [{'name': r[1], 'type': r[2].upper(), 'notnull': bool(r[3]), + 'default': r[4], 'pk': r[5]} + for r in connection.execute('PRAGMA table_info(' + quote(table) + ')')] + if any(c['type'] not in TYPES for c in columns): + raise ValueError('Unsupported storage type in ' + table) + pk = [c['name'] for c in sorted(columns, key=lambda c: c['pk']) if c['pk']] + if not pk: + raise ValueError('A deterministic primary key is required for ' + table) + indexes = [] + for row in connection.execute('PRAGMA index_list(' + quote(table) + ')'): + if row[3] == 'pk': + continue + if row[4]: + raise ValueError('Partial indexes require explicit migration support') + terms = [(r[2], bool(r[3]), r[4]) for r in connection.execute('PRAGMA index_xinfo(' + quote(row[1]) + ')') if r[5]] + if any(name is None or coll != 'BINARY' for name, desc, coll in terms): + raise ValueError('Unsupported expression or collation in ' + table) + indexes.append({'name': row[1], 'unique': bool(row[2]), 'origin': row[3], + 'terms': [(name, desc) for name, desc, coll in terms]}) + fks = {} + for row in connection.execute('PRAGMA foreign_key_list(' + quote(table) + ')'): + if row[5:7] != ('NO ACTION', 'NO ACTION'): + raise ValueError('Unsupported foreign key actions') + fk = fks.setdefault(row[0], {'table': row[2], 'columns': [], 'references': []}) + fk['columns'].append(row[3]); fk['references'].append(row[4]) + ddl = connection.execute("SELECT sql FROM sqlite_schema WHERE type='table' AND name=?", (table,)).fetchone()[0] + checks = CHECKS.get(table, []) + if sorted(map(normalize_check, checks_in(ddl))) != sorted(map(normalize_check, checks)): + raise ValueError('Unrecognized CHECK constraint in ' + table) + result[table] = {'columns': columns, 'pk': pk, 'indexes': indexes, + 'foreign_keys': list(fks.values()), 'checks': checks} + return result + + +def fingerprint(plan): + return hashlib.sha256(json.dumps(plan, sort_keys=True, separators=(',', ':')).encode()).hexdigest() + + +def default_sql(value): + if value is None: + return '' + if value.upper() == 'CURRENT_TIMESTAMP': + return " DEFAULT to_char(CURRENT_TIMESTAMP AT TIME ZONE 'UTC','YYYY-MM-DD\"T\"HH24:MI:SS.MS\"Z\"')" + if re.fullmatch(r'-?\d+|\'(?:[^\']|\'\')*\'', value): + return ' DEFAULT ' + value + raise ValueError('Unsupported column default') + + +def table_sql(schema, table, spec): + columns = [] + for column in spec['columns']: + generated = column['name'] == 'id' and spec['pk'] == ['id'] and column['type'] == 'INTEGER' + value = quote(column['name']) + ' ' + TYPES[column['type']] + value += ' GENERATED BY DEFAULT AS IDENTITY' if generated else default_sql(column['default']) + if column['notnull']: + value += ' NOT NULL' + columns.append(value) + return 'CREATE TABLE ' + quote(schema) + '.' + quote(table) + '(' + ','.join(columns) + ')' + + +def select_sql(schema, table, spec): + prefix = quote(schema) + '.' if schema else '' + return ('SELECT ' + ','.join(quote(c['name']) for c in spec['columns']) + + ' FROM ' + prefix + quote(table) + ' ORDER BY ' + ','.join(quote(c) for c in spec['pk'])) diff --git a/paper_atlas/querying.py b/paper_atlas/querying.py index f4af7dd..cc82979 100644 --- a/paper_atlas/querying.py +++ b/paper_atlas/querying.py @@ -23,8 +23,11 @@ def known_acceptances(conn,paper): def main(argv=None, *, prog=None): parser = argparse.ArgumentParser(prog=prog) - parser.add_argument('--db', required=True) - parser.add_argument('--query', help='SQLite FTS5 query over title and abstract') + database = parser.add_mutually_exclusive_group(required=True) + database.add_argument('--db', help='SQLite file') + database.add_argument('--pg-service', help='Protected libpq service; credentials stay outside CLI arguments') + parser.add_argument('--pg-schema', help='Optional PostgreSQL schema override; otherwise use the configured search_path') + parser.add_argument('--query', help='Title/abstract search: FTS5 for SQLite, web-search syntax for PostgreSQL') parser.add_argument('--paper',type=int,help='Paper ID; include sources, tags and submission history') parser.add_argument('--tag', help='Filter as namespace=value, e.g. arxiv:category=cs.AI') parser.add_argument('--venue', help='Conference acronym') @@ -35,9 +38,15 @@ def main(argv=None, *, prog=None): parser.add_argument('--limit', type=int, default=20) parser.add_argument('--stats', action='store_true') args = parser.parse_args(argv) - conn = sqlite3.connect(Path(args.db).resolve().as_uri()+'?mode=ro', uri=True) - conn.row_factory = sqlite3.Row - conn.execute('BEGIN') # One consistent read snapshot while ingestion continues. + if args.pg_schema and not args.pg_service: + parser.error('--pg-schema requires --pg-service') + if args.pg_service: + from paper_atlas.postgres.client import ReadConnection + conn = ReadConnection(args.pg_service, schema=args.pg_schema) if args.pg_schema else ReadConnection(args.pg_service) + else: + conn = sqlite3.connect(Path(args.db).resolve().as_uri()+'?mode=ro', uri=True) + conn.row_factory = sqlite3.Row + conn.execute('BEGIN') # One consistent read snapshot while ingestion continues. if args.stats: result = {name:conn.execute('SELECT count(*) FROM '+name).fetchone()[0] for name in ('papers','source_records','submissions','decisions','tags','match_candidates')} result['records_by_source'] = [dict(r) for r in conn.execute('SELECT s.name,count(r.id) AS imported_records FROM sources s LEFT JOIN source_records r ON r.source_id=s.id GROUP BY s.id ORDER BY s.name')] @@ -53,7 +62,12 @@ def main(argv=None, *, prog=None): if args.paper is not None: where.append('p.id=?'); params.append(args.paper) if args.query: - where.append('p.id IN (SELECT rowid FROM papers_fts WHERE papers_fts MATCH ?)'); params.append(args.query) + if args.pg_service: + from paper_atlas.postgres.client import SEARCH_VECTOR + where.append('p.id IN (SELECT id FROM papers WHERE '+SEARCH_VECTOR+" @@ websearch_to_tsquery('english'::regconfig,?))") + else: + where.append('p.id IN (SELECT rowid FROM papers_fts WHERE papers_fts MATCH ?)') + params.append(args.query) if args.tag: namespace,value = args.tag.split('=',1) where.append('EXISTS(SELECT 1 FROM paper_tags pt JOIN tags t ON t.id=pt.tag_id WHERE pt.paper_id=p.id AND t.namespace=? AND t.value=?)'); params.extend([namespace,value]) @@ -63,9 +77,9 @@ def main(argv=None, *, prog=None): clause = ['1=1'] if args.venue: if conn.execute("SELECT 1 FROM sqlite_master WHERE type='table' AND name='venue_aliases'").fetchone(): - clause.append('(v.acronym=? COLLATE NOCASE OR EXISTS(SELECT 1 FROM venue_aliases va WHERE va.venue_id=v.id AND va.alias=? COLLATE NOCASE))'); params.extend([args.venue,args.venue]) + clause.append('(lower(v.acronym)=lower(?) OR EXISTS(SELECT 1 FROM venue_aliases va WHERE va.venue_id=v.id AND lower(va.alias)=lower(?)))'); params.extend([args.venue,args.venue]) else: - clause.append('v.acronym=? COLLATE NOCASE');params.append(args.venue) + clause.append('lower(v.acronym)=lower(?)');params.append(args.venue) if args.track: clause.append('e.track=?');params.append(args.track) if args.year: diff --git a/pyproject.toml b/pyproject.toml index fdfb5af..0357a1c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -10,6 +10,9 @@ readme = "README.md" requires-python = ">=3.12" dependencies = ["PyYAML==6.0.3"] +[project.optional-dependencies] +postgres = ["psycopg[binary]==3.3.8"] + [project.scripts] paper-atlas = "paper_atlas.cli:main" diff --git a/tests/test_postgres.py b/tests/test_postgres.py new file mode 100644 index 0000000..523b604 --- /dev/null +++ b/tests/test_postgres.py @@ -0,0 +1,212 @@ +import contextlib +from contextlib import closing +import io +import json +import os +from pathlib import Path +import sqlite3 +import tempfile +import unittest +from unittest.mock import patch +import uuid + +from paper_atlas.db.store import Store +from paper_atlas.postgres import schema +from paper_atlas.postgres.client import ReadConnection, Row, connect +from paper_atlas.postgres.migration import migrate +from paper_atlas import querying + + +class SchemaTests(unittest.TestCase): + def setUp(self): + self.temporary = tempfile.TemporaryDirectory() + self.addCleanup(self.temporary.cleanup) + self.path = Path(self.temporary.name) / 'source.sqlite' + self.store = Store(self.path) + self.addCleanup(self.store.conn.close) + + def test_every_business_table_is_included_and_engine_tables_are_excluded(self): + plan = schema.inspect(self.store.conn) + self.assertEqual(set(plan), set(schema.TABLES)) + self.assertEqual(len(plan), 21) + self.assertNotIn('papers_fts', plan) + self.assertEqual(plan['paper_tags']['pk'], ['paper_id', 'tag_id', 'evidence_key']) + self.assertEqual(len(plan['source_records']['foreign_keys']), 2) + sql = schema.table_sql('stage', 'papers', plan['papers']) + self.assertIn('GENERATED BY DEFAULT AS IDENTITY', sql) + self.assertIn('TEXT COLLATE "C"', sql) + + def test_unknown_tables_and_check_constraints_fail_before_copy(self): + self.store.conn.execute('CREATE TABLE extra(id INTEGER PRIMARY KEY)') + with self.assertRaisesRegex(ValueError, 'Unexpected business'): + schema.inspect(self.store.conn) + self.store.conn.execute('DROP TABLE extra') + self.store.conn.execute('ALTER TABLE papers ADD COLUMN extra TEXT CHECK(extra IS NULL)') + with self.assertRaisesRegex(ValueError, 'CHECK'): + schema.inspect(self.store.conn) + + def test_cutover_stops_legacy_writers_but_keeps_read_access(self): + self.store.conn.commit(); self.store.conn.close() + Path(str(self.path) + '.postgresql-cutover.json').write_text('{"status":"active"}') + with self.assertRaisesRegex(RuntimeError, 'cut over'): + Store(self.path) + with sqlite3.connect(self.path.as_uri() + '?mode=ro', uri=True) as c: + self.assertEqual(c.execute('SELECT count(*) FROM papers').fetchone()[0], 0) + + def test_rows_support_mapping_and_positional_query_access(self): + row = Row(['id', 'title'], (7, 'A')) + self.assertEqual(row[0], row['id']) + self.assertEqual(dict(row), {'id': 7, 'title': 'A'}) + + def test_connection_services_cannot_be_password_urls(self): + with self.assertRaisesRegex(ValueError, 'service name'): + connect('postgresql://user:secret@host/db') + + def test_checks_compare_literal_values_not_only_constraint_counts(self): + clauses = schema.checks_in("CREATE TABLE t(x TEXT CHECK(x IN ('a','b)')),y TEXT CHECK(y IS NULL))") + self.assertEqual(clauses, ["x IN ('a','b)')", 'y IS NULL']) + self.assertEqual(schema.normalize_check("x IN ( 'a' )"), schema.normalize_check("x in('a')")) + self.assertNotEqual(schema.normalize_check("x IN ('A')"), schema.normalize_check("x IN ('a')")) + + +@unittest.skipUnless(os.environ.get('PAPER_ATLAS_TEST_PG_SERVICE'), 'Requires an explicitly configured remote PostgreSQL test service') +class PostgreSQLIntegrationTests(unittest.TestCase): + def setUp(self): + self.service = os.environ['PAPER_ATLAS_TEST_PG_SERVICE'] + self.target = 'pa_test_' + uuid.uuid4().hex[:14] + self.temporary = tempfile.TemporaryDirectory() + self.addCleanup(self.temporary.cleanup) + self.db = Path(self.temporary.name) / 'source.sqlite' + self.out = Path(self.temporary.name) / 'report.json' + store = Store(self.db) + paper, record = store.record('arxiv', '2401.00001', 'https://arxiv.org/abs/2401.00001', + 'Quiet graph models — 研究', 'Graph models learn reinforcement policies.\nQuoted "text", tabs\tand Unicode α.', + ['Jane Doe', 'John Smith'], {'arxiv': '2401.00001'}, {'nested': {'a': ['α', None, 1]}, 'raw': '\\n'}) + store.record_tags(paper, record, 'arxiv', [{'namespace': 'arxiv:category', 'value': 'cs.AI'}]) + store.decision(paper, record, 'ICLR', 2024, 'main', 'forum-reject', 'https://example.org/reject', + 'rejected', 'Reject', identities=[{'scheme': 'openreview', 'value': 'forum-reject'}], observed_at='2024-01-01T00:00:00Z') + same, second = store.record('official', '2025/paper', 'https://example.org/accepted', + 'Quiet graph models — 研究', 'Graph models learn reinforcement policies.', ['Jane Doe', 'John Smith'], {'arxiv': '2401.00001'}) + self.assertEqual(same, paper) + store.decision(paper, second, 'COLM', 2025, 'main', 'forum-accept', 'https://example.org/accept', + 'accepted', 'Accept', ['Conference on Language Modeling'], observed_at='2025-01-01T00:00:00Z') + store.record('arxiv', '2401.00002', 'https://arxiv.org/abs/2401.00002', 'Empty abstract', None, []) + tag = store.conn.execute("INSERT INTO tags(namespace,value) VALUES('custom:topic','manual')").lastrowid + store.conn.execute('INSERT INTO paper_tags VALUES(?,?,NULL,?)', (paper, tag, 'manual:one')) + store.conn.execute("INSERT INTO paper_merges(former_paper_id,surviving_paper_id,reason) VALUES(900,?,'fixture')", (paper,)) + store.conn.execute("INSERT INTO submission_merges VALUES(5000,1,'fixture',CURRENT_TIMESTAMP)") + store.conn.execute("INSERT INTO decision_merges VALUES(6000,1,'fixture',CURRENT_TIMESTAMP)") + store.conn.execute("UPDATE sqlite_sequence SET seq=1000 WHERE name='papers'") + store.conn.commit(); store.conn.close() + self.addCleanup(self.cleanup_schemas) + + def cleanup_schemas(self): + with connect(self.service) as pg: + names = pg.execute('SELECT nspname FROM pg_namespace WHERE starts_with(nspname,%s)', (self.target,)).fetchall() + for name, in names: + pg.execute('DROP SCHEMA ' + schema.quote(name) + ' CASCADE') + + def migrate(self): + with contextlib.redirect_stdout(io.StringIO()): + return migrate(self.db, self.service, self.out, schema=self.target, min_free_gib=0) + + def published(self): + with connect(self.service) as pg: + return bool(pg.execute('SELECT 1 FROM pg_namespace WHERE nspname=%s', (self.target,)).fetchone()) + + def test_full_content_history_tags_sequences_and_readonly_queries(self): + report = self.migrate() + self.assertEqual(report['status'], 'complete') + self.assertEqual(len(report['tables']), 21) + for entry in report['tables'].values(): + self.assertEqual(entry['source_rows'], entry['target_rows']) + self.assertEqual(entry['source_sha256'], entry['target_sha256']) + with connect(self.service) as pg, pg.transaction(force_rollback=True): + prefix = schema.quote(self.target) + '.' + value = pg.execute('INSERT INTO ' + prefix + 'papers(title,title_key) VALUES(%s,%s) RETURNING id', ('new', 'new')).fetchone()[0] + self.assertGreater(value, 1000) + record = pg.execute('INSERT INTO ' + prefix + "source_records(source_id,external_id,paper_id,url,title,abstract,authors_json,metadata_json) SELECT source_id,'new',%s,url,title,abstract,authors_json,metadata_json FROM " + prefix + 'source_records LIMIT 1 RETURNING id', (value,)).fetchone()[0] + edition = pg.execute('SELECT id FROM ' + prefix + 'editions LIMIT 1').fetchone()[0] + submission = pg.execute('INSERT INTO ' + prefix + 'submissions(paper_id,edition_id,source_record_id,source_submission_id) VALUES(%s,%s,%s,%s) RETURNING id', (value, edition, record, 'new')).fetchone()[0] + decision = pg.execute('INSERT INTO ' + prefix + "decisions(submission_id,status,raw_value,evidence_url,observed_at) VALUES(%s,'unknown','fixture','https://example.org/new','2026-01-01T00:00:00Z') RETURNING id", (submission,)).fetchone()[0] + self.assertGreater(submission, 5000) + self.assertGreater(decision, 6000) + real = ReadConnection + with patch('paper_atlas.postgres.client.ReadConnection', lambda service: real(service, self.target)): + output = io.StringIO() + with contextlib.redirect_stdout(output): + querying.main(['--pg-service', self.service, '--query', 'graph models', '--status', 'rejected', '--outcome', 'accepted']) + results = json.loads(output.getvalue()) + self.assertEqual(len(results), 1) + self.assertEqual(results[0]['known_acceptances'][0]['venue'], 'COLM') + output = io.StringIO() + with contextlib.redirect_stdout(output): + querying.main(['--pg-service', self.service, '--venue', 'conference on language modeling', '--tag', 'custom:topic=manual', '--paper', '1']) + detail = json.loads(output.getvalue())[0] + self.assertEqual(len(detail['submissions']), 2) + self.assertEqual({t['namespace'] for t in detail['tags']}, {'arxiv:category', 'custom:topic'}) + output = io.StringIO() + with contextlib.redirect_stdout(output): + querying.main(['--pg-service', self.service, '--stats']) + stats = json.loads(output.getvalue()) + self.assertEqual(stats['papers'], 2) + self.assertEqual(stats['papers_with_known_acceptance'], 1) + reader = real(self.service, self.target) + try: + import psycopg + with self.assertRaises(psycopg.errors.ReadOnlySqlTransaction): + reader.execute('DELETE FROM papers') + finally: + reader.close() + + def interrupt_after_first_table(self): + calls = 0 + def interrupt(*args): + nonlocal calls + calls += 1 + if calls == 3: + raise RuntimeError('controlled interruption') + with patch('paper_atlas.postgres.migration.free_space', side_effect=interrupt): + with self.assertRaisesRegex(RuntimeError, 'controlled interruption'): + self.migrate() + self.assertFalse(self.published()) + + def test_interrupted_copy_resumes_without_duplicate_rows(self): + self.interrupt_after_first_table() + self.assertEqual(self.migrate()['status'], 'complete') + + def test_resume_rejects_a_changed_source_snapshot(self): + self.interrupt_after_first_table() + with closing(sqlite3.connect(self.db)) as source, source: + source.execute("UPDATE papers SET abstract='changed' WHERE id=1") + with self.assertRaisesRegex(ValueError, 'exact same'): + self.migrate() + self.assertFalse(self.published()) + + def test_full_hash_detects_tampering_in_a_completed_staging_table(self): + self.interrupt_after_first_table() + with connect(self.service) as pg: + stage = pg.execute('SELECT stage FROM ' + schema.quote(self.target + '_migration') + '.run WHERE id=1').fetchone()[0] + pg.execute('UPDATE ' + schema.quote(stage) + ".sources SET homepage='tampered'") + with self.assertRaisesRegex(ValueError, 'content verification'): + self.migrate() + self.assertFalse(self.published()) + + def test_foreign_key_violation_never_publishes_target(self): + with closing(sqlite3.connect(self.db)) as source, source: + source.execute('PRAGMA foreign_keys=OFF') + source.execute("INSERT INTO identifiers VALUES('orphan','bad',999999)") + import psycopg + with self.assertRaises(psycopg.errors.ForeignKeyViolation): + self.migrate() + self.assertFalse(self.published()) + + def test_published_target_cannot_be_overwritten(self): + self.migrate() + self.out.unlink() + with self.assertRaisesRegex(ValueError, 'never be overwritten'): + self.migrate() + + +if __name__ == '__main__': + unittest.main() -- 2.52.0