Preserve NUL text through PostgreSQL migration and resume

Reviewed-on: #13

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`，其他字符不变；包括字串前缀在内的二进制排序保持一致，主键、唯一约束和外键可继续验证。编码版本保存在 `<schema>_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 只读备份，并在旁保存 `<SQLite路径>.postgresql-cutover.json`，记录报告路径、源哈希、目标 service/schema；新版本旧写入入口见此标记会拒绝写入。标记不含密码。后续导入适配属于下一阶段，不可继续更新 SQLite 后声称 PostgreSQL 已同步。

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 当前状态。
 

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

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)',
+                    pg.execute('INSERT INTO ' + qualify(control, 'run') +
-                               (json.dumps(identity, sort_keys=True), stage, 'copying'))
+                               '(id,source_json,stage,status,text_codec) VALUES(1,%s,%s,%s,%s)',
-            report.update(status='copying', stage=stage); save(out, report)
+                               (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())

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)

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()