diff --git a/docs/operations.md b/docs/operations.md index 6ac8f5a..766c8ba 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -18,8 +18,12 @@ paper-atlas maintain postgres \ 全部 21 张业务表保留原 ID、原始 JSON 文本、NULL、来源、投稿、决定、标签出处和合并记录。`sqlite_sequence` 用于重设 PostgreSQL identity 序列，并考虑已合并的历史 ID，避免复用；SQLite FTS5 shadow 表根据完整标题/摘要重建为原生 GIN 索引，不逐字复制引擎文件。 +SQLite 允许文本包含 `U+0000`（NUL），PostgreSQL `text` 不允许。迁移使用 `nul-soh-v1` 可逆编码：NUL 存为 `U+0001 U+0001`，原有 `U+0001` 存为 `U+0001 U+0002`，其他字符不变；包括字串前缀在内的二进制排序保持一致，主键、唯一约束和外键可继续验证。编码版本保存在 `_migration.run.text_codec`，发布时同时写入 schema 注释 `paper-atlas:text-codec=nul-soh-v1`，其他项目只需研究表的权限即可读取格式信息，无须访问迁移控制表。CLI 的只读连接自动还原结果并编码筛选参数；直接 SQL/ETL 调用者应使用 `paper_atlas.postgres.text.encode/decode`，不能删除或直接替换控制字符。全文检索把这些控制字符当作分隔符，不承担逐字节匹配。 + 先 COPY 到独立暂存 schema，每张表提交和记录哈希。中断后以同一命令、同一报告路径恢复，必须仍是**完全相同的未变更快照**。不同输入、已发布目标、未清空 WAL、新的未知业务表都会拒绝覆盖；恢复只会清空本次运行尚未完成且尚未发布的暂存表。保留错误现场，不盲目从头复制。 +恢复旧版本未发布的暂存数据时，在一个事务中转义已提交表里原有的 `U+0001`，然后记录新编码版本；失败回滚，重复恢复不会二次编码。已经提交的表无需从 SQLite 重新复制。报告中的 `source_sha256`、`target_sha256` 比较完整还原后的原始值；`target_storage_sha256` 另记实际存储值的哈希，包含转义的表两种哈希可以不同。 + 发布前逐表按主键完整回读，比较数量及逐行 SHA-256；建立并验证所有主键、唯一约束、CHECK 和外键，重设序列、建立 GIN 并 ANALYZE。全部通过后原子重命名为 `research` schema，报告才为 `complete`。`copying`、`prepared`、部分表通过或 CI 通过均不代表生产迁移完成，也不替代原任务最后的实体去重和会议覆盖审计。 发布后核对 PostgreSQL 的统计、全文检索、论文详情，以及 reject 后另会 accept 的历史。确认后保留 SQLite 只读备份，并在旁保存 `.postgresql-cutover.json`，记录报告路径、源哈希、目标 service/schema；新版本旧写入入口见此标记会拒绝写入。标记不含密码。后续导入适配属于下一阶段，不可继续更新 SQLite 后声称 PostgreSQL 已同步。 diff --git a/docs/usage.md b/docs/usage.md index 12503f6..c8d9895 100644 --- a/docs/usage.md +++ b/docs/usage.md @@ -14,7 +14,9 @@ paper-atlas query --pg-service foundry-research --status rejected --outcome acce 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`。 +使用只读一致性事务，支持论文 ID、会议别名、年份、轨道、标签、最新决定和跨会议录用筛选。全文检索使用 PostgreSQL `websearch_to_tsquery('english', ...)`，支持引号短语、`OR` 和减号排除；与 SQLite FTS5 的语法及分词结果不完全相同。JSON 来源列保留原始文本；转成 `jsonb` 前须确认原始值符合该类型的限制。 + +CLI 自动还原迁移中的 NUL/SOH 文本转义，详情中的来源摘要和筛选参数保留原字符。直接 SQL 读取可能看到 `U+0001` 转义序列；跨项目 ETL 应按迁移元数据中的 `text_codec` 使用 `paper_atlas.postgres.text.encode/decode`。存储格式和无损哈希验收见[运行维护](operations.md#postgresql-迁移)。 迁移与切换见[运行维护](operations.md#postgresql-迁移)。下述 SQLite 导入命令仅用于未切换的索引或独立样本；切换后的生产源拒绝旧写入。尚未适配的覆盖审计可只读检查冻结备份，不能当作 PostgreSQL 当前状态。 diff --git a/paper_atlas/postgres/client.py b/paper_atlas/postgres/client.py index 076c6c5..8c8e784 100644 --- a/paper_atlas/postgres/client.py +++ b/paper_atlas/postgres/client.py @@ -1,6 +1,7 @@ """Libpq service connections; credentials never belong in CLI arguments.""" from collections.abc import Mapping import re +from paper_atlas.postgres import text SEARCH_VECTOR = "to_tsvector('english'::regconfig,coalesce(title,'') || ' ' || coalesce(abstract,''))" @@ -44,9 +45,9 @@ class Row(Mapping): return len(self.columns) -def row_factory(cursor): +def row_factory(cursor, *, encoded=False): columns = [column.name for column in cursor.description] if cursor.description else [] - return lambda values: Row(columns, values) + return lambda values: Row(columns, tuple(map(text.decode, values)) if encoded else values) class ReadConnection: @@ -59,10 +60,21 @@ class ReadConnection: 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') + current = self.connection.execute('SELECT current_schema()').fetchone()[0] + # Schema comments are readable by other projects granted table access; + # consumers need no privileges on the private migration controller. + annotation = self.connection.execute("SELECT obj_description(oid,'pg_namespace') FROM pg_namespace WHERE nspname=%s", (current,)).fetchone()[0] + self.encoded = False + if annotation and annotation.startswith('paper-atlas:text-codec='): + codec = annotation.removeprefix('paper-atlas:text-codec=') + if codec != text.CODEC: + self.close() + raise RuntimeError('Unsupported PostgreSQL text storage codec') + self.encoded = codec == text.CODEC + self.connection.row_factory = lambda cursor: row_factory(cursor, encoded=self.encoded) def execute(self, query, params=()): # All caller SQL is static; user values are always bound parameters. @@ -73,7 +85,8 @@ class ReadConnection: 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) + values = tuple(map(text.encode, params)) if self.encoded else params + return self.connection.execute(query.replace('?', '%s'), values) def has_table(self, name): return bool(self.connection.execute('''SELECT 1 FROM information_schema.tables diff --git a/paper_atlas/postgres/migration.py b/paper_atlas/postgres/migration.py index be97f52..0a051ed 100644 --- a/paper_atlas/postgres/migration.py +++ b/paper_atlas/postgres/migration.py @@ -14,6 +14,7 @@ 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 +from paper_atlas.postgres import text def digest_file(path): @@ -96,6 +97,41 @@ def sequences(pg, source, stage, plan): return result +def upgrade_text_codec(pg, control, stage, plan): + """Upgrade committed legacy staging rows once; never recopy their source.""" + changed = {} + with pg.transaction(): + copied = {r[0] for r in pg.execute('SELECT name FROM ' + qualify(control, 'tables'))} + if not copied <= plan.keys(): + raise ValueError('Unknown committed staging tables') + # An older run may have reached constraint validation. Replacing SOH + # can temporarily equal another legacy key even though the final mapping + # is bijective. Rebuild unpublished constraints/indexes after the upgrade. + 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 co.contype IN ('f','p','u') + ORDER BY CASE WHEN co.contype='f' THEN 0 ELSE 1 END''', (stage,)).fetchall() + for table, name in constraints: + pg.execute('ALTER TABLE ' + qualify(stage, table) + ' DROP CONSTRAINT ' + quote(name)) + indexes = pg.execute('''SELECT cl.relname FROM pg_index i JOIN pg_class cl ON cl.oid=i.indexrelid + JOIN pg_namespace ns ON ns.oid=cl.relnamespace WHERE ns.nspname=%s AND i.indisunique''', (stage,)).fetchall() + for name, in indexes: + pg.execute('DROP INDEX ' + qualify(stage, name)) + for table in plan: + if table not in copied: + continue + columns = [quote(c['name']) for c in plan[table]['columns'] if c['type'] == 'TEXT'] + if columns: + assignments = ','.join(c + '=replace(' + c + ',%s,%s)' for c in columns) + predicate = ' OR '.join('strpos(' + c + ',%s)>0' for c in columns) + params = [v for c in columns for v in (text.ESCAPE, text.encode(text.ESCAPE))] + params += [text.ESCAPE] * len(columns) + changed[table] = pg.execute('UPDATE ' + qualify(stage, table) + ' SET ' + assignments + + ' WHERE ' + predicate, params).rowcount + pg.execute('UPDATE ' + qualify(control, 'run') + ' SET text_codec=%s WHERE id=1', (text.CODEC,)) + return changed + + 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') @@ -129,20 +165,28 @@ def migrate(db, service, out, *, schema='research', storage_path=None, min_free_ '(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() + pg.execute('ALTER TABLE ' + qualify(control, 'run') + " ADD COLUMN IF NOT EXISTS text_codec TEXT NOT NULL DEFAULT 'legacy'") + run = pg.execute('SELECT source_json,stage,status,text_codec 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] + if run[3] == 'legacy': + report['legacy_text_upgrade_rows'] = upgrade_text_codec(pg, control, stage, plan) + elif run[3] != text.CODEC: + raise ValueError('Unsupported staging text storage codec') 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) + pg.execute('INSERT INTO ' + qualify(control, 'run') + + '(id,source_json,stage,status,text_codec) VALUES(1,%s,%s,%s,%s)', + (json.dumps(identity, sort_keys=True), stage, 'copying', text.CODEC)) + report.update(status='copying', stage=stage, text_codec=text.CODEC, + content_hash_scope='Decoded original SQLite values; storage escapes are reversible.') + save(out, report) try: for table, spec in plan.items(): free_space(storage_path, min_free_gib) @@ -157,7 +201,7 @@ def migrate(db, service, out, *, schema='research', storage_path=None, min_free_ 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 + copy.write_row(tuple(map(text.encode, row))); update_hash(digest, row); rows += 1 if rows % 10000 == 0: free_space(storage_path, min_free_gib) if rows % 100000 == 0: @@ -172,14 +216,16 @@ def migrate(db, service, out, *, schema='research', storage_path=None, min_free_ 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 + digest = hashlib.sha256(); storage_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 + update_hash(storage_digest, row) + update_hash(digest, tuple(map(text.decode, row))); rows += 1 entry = report['tables'][table] - entry.update(target_rows=rows, target_sha256=digest.hexdigest()) + entry.update(target_rows=rows, target_sha256=digest.hexdigest(), + target_storage_sha256=storage_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,)) @@ -207,6 +253,7 @@ def migrate(db, service, out, *, schema='research', storage_path=None, min_free_ report.update(status='prepared', foreign_keys='all_validated', checks='all_validated', native_search_index=True) save(out, report) with pg.transaction(): + pg.execute('COMMENT ON SCHEMA ' + quote(stage) + " IS 'paper-atlas:text-codec=" + text.CODEC + "'") 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()) diff --git a/paper_atlas/postgres/text.py b/paper_atlas/postgres/text.py new file mode 100644 index 0000000..f38fbcb --- /dev/null +++ b/paper_atlas/postgres/text.py @@ -0,0 +1,31 @@ +"""Order-preserving, reversible storage for SQLite text containing NUL. + +PostgreSQL text forbids U+0000. Encode it as SOH/SOH, and encode a literal +SOH as SOH/STX. All other characters stay unchanged. The mapping preserves +binary lexicographic order, including string prefixes and composite keys. +""" + +CODEC = 'nul-soh-v1' +ESCAPE = '\x01' + + +def encode(value): + if isinstance(value, str): + return value.replace(ESCAPE, ESCAPE + '\x02').replace('\x00', ESCAPE + ESCAPE) + return value + + +def decode(value): + if not isinstance(value, str) or ESCAPE not in value: + return value + result = [] + start = 0 + while (position := value.find(ESCAPE, start)) >= 0: + result.append(value[start:position]) + following = value[position + 1:position + 2] + if following not in (ESCAPE, '\x02'): + raise ValueError('Invalid nul-soh-v1 text escape') + result.append('\x00' if following == ESCAPE else ESCAPE) + start = position + 2 + result.append(value[start:]) + return ''.join(result) diff --git a/tests/test_postgres.py b/tests/test_postgres.py index 523b604..36c383b 100644 --- a/tests/test_postgres.py +++ b/tests/test_postgres.py @@ -1,6 +1,7 @@ import contextlib from contextlib import closing import io +import itertools import json import os from pathlib import Path @@ -11,7 +12,7 @@ from unittest.mock import patch import uuid from paper_atlas.db.store import Store -from paper_atlas.postgres import schema +from paper_atlas.postgres import schema, text from paper_atlas.postgres.client import ReadConnection, Row, connect from paper_atlas.postgres.migration import migrate from paper_atlas import querying @@ -68,6 +69,21 @@ class SchemaTests(unittest.TestCase): 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')")) + def test_text_codec_is_bijective_and_preserves_key_order_and_prefixes(self): + alphabet = ['\x00', '\x01', '\x02', 'a', 'é', '🙂'] + values = [''.join(v) for n in range(4) for v in itertools.product(alphabet, repeat=n)] + for value in values: + encoded = text.encode(value) + self.assertNotIn('\x00', encoded) + self.assertEqual(text.decode(encoded), value) + self.assertEqual(sorted(map(text.encode, values)), list(map(text.encode, sorted(values)))) + for value in [None, 7, 0.5]: + self.assertEqual(text.encode(value), value) + self.assertEqual(text.decode(value), value) + for value in ['\x01', 'prefix\x01x']: + with self.assertRaisesRegex(ValueError, 'Invalid'): + text.decode(value) + @unittest.skipUnless(os.environ.get('PAPER_ATLAS_TEST_PG_SERVICE'), 'Requires an explicitly configured remote PostgreSQL test service') class PostgreSQLIntegrationTests(unittest.TestCase): @@ -207,6 +223,87 @@ class PostgreSQLIntegrationTests(unittest.TestCase): with self.assertRaisesRegex(ValueError, 'never be overwritten'): self.migrate() + def test_nul_and_literal_escape_survive_full_copy_keys_query_and_hash(self): + original = 'before\x00survivor; literal\x01\x01 and \x01\x02 — 研究' + keys = ['a\x00z', 'a\x01z', 'a\x01\x02z', 'a\x02z'] + with closing(sqlite3.connect(self.db)) as source, source: + source.execute('UPDATE papers SET abstract=? WHERE id=1', (original,)) + source.execute('UPDATE source_records SET abstract=? WHERE id=1', (original,)) + source.executemany("INSERT INTO identifiers VALUES('control',?,1)", [(k,) for k in keys]) + source.execute("INSERT INTO tags(namespace,value) VALUES('custom:control',?)", (original,)) + tag = source.execute("SELECT id FROM tags WHERE namespace='custom:control'").fetchone()[0] + source.executemany('INSERT INTO paper_tags VALUES(1,?,NULL,?)', [(tag, k) for k in keys]) + report = self.migrate() + self.assertEqual(report['text_codec'], text.CODEC) + for entry in report['tables'].values(): + self.assertEqual(entry['source_sha256'], entry['target_sha256']) + self.assertNotEqual(report['tables']['source_records']['source_sha256'], + report['tables']['source_records']['target_storage_sha256']) + with connect(self.service) as pg: + self.assertEqual(pg.execute("SELECT obj_description(oid,'pg_namespace') FROM pg_namespace WHERE nspname=%s", (self.target,)).fetchone()[0], 'paper-atlas:text-codec=' + text.CODEC) + raw = pg.execute('SELECT abstract FROM ' + schema.quote(self.target) + '.source_records WHERE id=1').fetchone()[0] + self.assertNotIn('\x00', raw) + self.assertEqual(text.decode(raw), original) + reader = ReadConnection(self.service, self.target) + try: + self.assertEqual(reader.execute('SELECT abstract FROM source_records WHERE id=?', (1,)).fetchone()[0], original) + self.assertEqual([r[0] for r in reader.execute("SELECT value FROM identifiers WHERE scheme='control' ORDER BY value")], sorted(keys)) + self.assertEqual(reader.execute("SELECT value FROM identifiers WHERE scheme='control' AND value=?", (keys[0],)).fetchone()[0], keys[0]) + finally: + reader.close() + 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, '--paper', '1', '--tag', 'custom:control=' + original]) + result = json.loads(output.getvalue())[0] + self.assertEqual(result['abstract'], original) + self.assertEqual(next(s['abstract'] for s in result['sources'] if s['external_id'] == '2401.00001'), original) + output = io.StringIO() + with contextlib.redirect_stdout(output): + querying.main(['--pg-service', self.service, '--query', 'survivor']) + self.assertEqual([r['id'] for r in json.loads(output.getvalue())], [1]) + + def test_old_completed_staging_table_is_upgraded_atomically_on_resume(self): + original = 'https://example.org/literal\x01\x02' + with closing(sqlite3.connect(self.db)) as source, source: + source.execute('UPDATE sources SET homepage=? WHERE id=1', (original,)) + self.interrupt_after_first_table() + with connect(self.service) as pg: + control = schema.quote(self.target + '_migration') + stage = pg.execute('SELECT stage FROM ' + control + '.run WHERE id=1').fetchone()[0] + # Reproduce the old implementation: raw SOH text and no codec metadata. + pg.execute('UPDATE ' + schema.quote(stage) + '.sources SET homepage=%s WHERE id=1', (original,)) + pg.execute('ALTER TABLE ' + control + '.run DROP COLUMN text_codec') + report = self.migrate() + self.assertEqual(report['legacy_text_upgrade_rows']['sources'], 1) + self.assertEqual(report['tables']['sources']['source_sha256'], report['tables']['sources']['target_sha256']) + reader = ReadConnection(self.service, self.target) + try: + self.assertEqual(reader.execute('SELECT homepage FROM sources WHERE id=1').fetchone()[0], original) + finally: + reader.close() + + def test_legacy_unique_keys_can_upgrade_after_constraints_were_built(self): + keys = ['x\x01', 'x\x01\x02'] + with closing(sqlite3.connect(self.db)) as source, source: + source.executemany("INSERT INTO identifiers VALUES('control',?,1)", [(k,) for k in keys]) + # Emulate an old run interrupted after constraints: no storage encoding. + with patch('paper_atlas.postgres.text.encode', side_effect=lambda value: value), \ + patch('paper_atlas.postgres.migration.sequences', side_effect=RuntimeError('after constraints')): + with self.assertRaisesRegex(RuntimeError, 'after constraints'): + self.migrate() + with connect(self.service) as pg: + pg.execute('ALTER TABLE ' + schema.quote(self.target + '_migration') + '.run DROP COLUMN text_codec') + report = self.migrate() + self.assertEqual(report['status'], 'complete') + self.assertEqual(report['tables']['identifiers']['source_sha256'], report['tables']['identifiers']['target_sha256']) + reader = ReadConnection(self.service, self.target) + try: + self.assertEqual([r[0] for r in reader.execute("SELECT value FROM identifiers WHERE scheme='control' ORDER BY value")], keys) + finally: + reader.close() + if __name__ == '__main__': unittest.main()