perf(archive): batch target compatibility checks

docs/en/docs/operate/portable-archive.md

@@ -237,6 +237,56 @@ Logical export is not a substitute for a crash-consistent deployment backup:
 
 ## Current boundaries
 
+Target compatibility checks group records by type, with at most 100 records per SQL statement, 900 bound
+parameters across the pending groups, and 1 MiB of encoded lookup keys and expected digests. Wider composite
+keys can produce smaller batches. Pending groups retain keys and digests, not record bodies. The result cursor
+fetches one target row at a time; peak memory still includes a single decoded archive record, a target row,
+canonical encoding buffers, and database-driver overhead. A large existing target body therefore does not
+multiply by the batch size. This is a working-set bound, not a fixed process-memory limit.
+
+Each batch preserves the target database's key comparison rules and checks both logical identities and
+alternate physical unique keys. Any differing match makes that archive record a conflict, counted once.
+Nullable unique keys retain SQL null semantics. Dry-run results are not cached for a later restore: restore
+checks the current target again, and write-time conflicts still roll back the transaction. Publication-chain
+and combined-tag-set checks run after the record checks.
+
+Target-check progress is reported after each completed batch, even when its processed count skips a
+1,000-record reporting threshold. Records whose target tables are absent retain periodic per-record reporting.
+
+For the representative archive containing one Scope, one journal head, and N small Sources, the compatibility
+query budget against initialized tables is `ceil(N / 100) + 2`: 3, 12, and 52 statements for 100, 1,000, and
+5,000 Sources. Missing tables require no record compatibility queries. Schema introspection, local dependency
+index queries, and publication/tag integrity queries are separate from this budget.
+
+Run `uv run python scripts/benchmark_archive_validation.py --sources 100 1000 5000 --runs 3` to measure full
+validation times and query counts on disposable file-backed SQLite targets. Fixture generation and export
+are excluded. The separately reported restore duration includes revalidation, staging, authoritative writes,
+and the receipt, with no projection rebuilder configured; it is not an isolated write or search-rebuild timing.
+Compare the same script and environment across checkouts. Fewer SQL executions do not imply the same reduction
+in complete validation time: archive parsing, dependency checks, decoding, and digests remain per-record work.
+Embedded SeekDB is local; remote OceanBase costs depend on the deployment topology and need real-backend measurements.
+
+On 2026-10-23, the script was run against baseline `4a21da08` (PR #1561 plus current master) and the batched
+implementation on Windows 11 build 26200, CPython 3.12.13, SQLite 3.50.4, with default SQLiteProfile settings,
+256-byte Source content, and no injected network delay. Each cell below is the median of three full validations;
+query counts were identical in all repetitions and exclude schema introspection.
+
+| Records | Target | Queries: baseline → batched | Seconds: baseline → batched |
+| ---: | --- | ---: | ---: |
+| 102 | Empty | 102 → 3 | 0.3469 → 0.2828 |
+| 102 | Identical | 102 → 3 | 0.3755 → 0.3039 |
+| 1,002 | Empty | 1,002 → 12 | 2.6401 → 1.9954 |
+| 1,002 | Identical | 1,002 → 12 | 3.2584 → 2.3064 |
+| 5,002 | Empty | 5,002 → 52 | 16.3445 → 9.1713 |
+| 5,002 | Identical | 5,002 → 52 | 16.9955 → 10.7552 |
+| 5,002 | Tables absent | 0 → 0 | 7.9935 → 7.5444 |
+
+The missing-table control also varies despite executing no compatibility queries; these are local diagnostic
+measurements, not a throughput guarantee. One 5,002-record restore without projections took 17.6685 seconds on
+the baseline and 10.5197 seconds with batching, including internal validation and staging. Projection rebuilding
+was not measured. Native SeekDB and OceanBase were not executed in this environment; their parameterized archive
+acceptance tests require actual backend runs before those configurations are qualified for this optimization.
+
 Export streams database rows to a temporary NDJSON file. Validation stores record identities and Candidate versions in a
 temporary on-disk index to verify each head against its exact version. Restore replays dependency levels as streams, so
 aggregate archive payloads are not retained in

docs/zh/docs/operate/portable-archive.md

@@ -205,6 +205,46 @@ reader 会拒绝这些记录不完整的 schema version `1` 表示，而不是
 
 ## 当前边界
 
+目标兼容性检查按记录类型分组，每条 SQL 最多检查 100 条记录。待处理分组总计最多使用 900 个绑定参数，
+编码后的查询键和预期摘要总计最多 1 MiB；较宽的复合键会使批次提前拆分。缓存只保留键和摘要，不保留记录正文。
+结果游标每次读取一条目标记录，内存峰值仍包含单条已解码归档记录、单条目标记录、规范编码缓冲和数据库驱动开销。
+因此，目标中的大正文不会按批次条数成倍驻留内存。这是工作集边界，不是固定的进程内存上限。
+
+批量查询保留目标数据库的键比较规则，同时检查逻辑身份和其他物理唯一键。任何命中项内容不同，都将该归档记录
+计为一次冲突；可空唯一键继续遵循 SQL 的 null 语义。dry-run 结果不会缓存给后续恢复使用：restore 会重新检查
+当前目标，写入阶段的冲突仍会回滚整个事务。记录检查完成后，继续执行发布链和合并标签集合的完整性检查。
+
+目标检查每完成一批就报告一次进度，即使累计记录数跨过了 1,000 条记录的报告阈值。目标表不存在的记录仍按逐条处理的周期报告进度。
+
+对于一个 Scope、一个日志头和 N 条小 Source 组成的代表性归档，目标表已初始化时，兼容性查询预算为
+`ceil(N / 100) + 2`：100、1,000、5,000 条 Source 分别最多执行 3、12、52 条查询。
+目标表不存在时，不执行对应记录的兼容性查询。表结构探测、本地依赖索引查询、发布链和标签完整性查询单独计算。
+
+运行 `uv run python scripts/benchmark_archive_validation.py --sources 100 1000 5000 --runs 3`，
+可在临时文件型 SQLite 数据库上测量完整校验耗时和查询次数。计时不包含数据准备和导出。
+单独报告的恢复耗时包含重新校验、记录暂存、权威数据写入和回执，未配置投影重建器，不能解释成纯写入或搜索重建耗时。
+跨提交比较时使用同一脚本和环境。SQL 执行次数减少不代表完整校验耗时同比下降，归档解析、依赖检查、解码和摘要计算
+仍需要逐条完成。嵌入式 SeekDB 在本地运行；远程 OceanBase 的成本与部署拓扑有关，需要在真实后端上测量。
+
+2026-10-23 在 Windows 11 build 26200、CPython 3.12.13、SQLite 3.50.4 上，对基线 `4a21da08`
+（PR #1561 与当前 master 的合并代码）和批处理实现运行上述脚本。使用默认 SQLiteProfile 配置，Source 正文为
+256 字节，未注入网络延迟。以下耗时为三次完整校验的中位数，每次查询计数一致，不包含表结构探测。
+
+| 记录数 | 目标状态 | 查询数：基线 → 批处理 | 秒：基线 → 批处理 |
+| ---: | --- | ---: | ---: |
+| 102 | 空表 | 102 → 3 | 0.3469 → 0.2828 |
+| 102 | 已有相同数据 | 102 → 3 | 0.3755 → 0.3039 |
+| 1,002 | 空表 | 1,002 → 12 | 2.6401 → 1.9954 |
+| 1,002 | 已有相同数据 | 1,002 → 12 | 3.2584 → 2.3064 |
+| 5,002 | 空表 | 5,002 → 52 | 16.3445 → 9.1713 |
+| 5,002 | 已有相同数据 | 5,002 → 52 | 16.9955 → 10.7552 |
+| 5,002 | 表不存在 | 0 → 0 | 7.9935 → 7.5444 |
+
+不执行兼容性查询的缺表对照也有耗时波动，因此这些本地诊断数据不能作为吞吐量保证。
+5,002 条记录的一次不含投影重建的恢复，基线耗时 17.6685 秒，批处理耗时 10.5197 秒，均包含内部校验和暂存。
+本次未测量投影重建，也未在该环境执行原生 SeekDB 或 OceanBase。针对这些配置，仍需运行参数化归档验收测试，
+取得真实后端执行结果后才能确认本优化的适用性。
+
 导出会把数据库 row 流式写到临时 NDJSON 文件。验证在临时磁盘 index 中保存 record identity 和 Candidate version，
 用于校验每个 head 对应的精确版本。恢复则按依赖层流式
 回放，因此归档的聚合 payload 不会常驻进程内存；Scope metadata 由 manifest scope list 约束。命令不是远程 Server

scripts/benchmark_archive_validation.py

@@ -0,0 +1,164 @@
+# Copyright (c) 2026 OceanBase.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Measure public archive validation on disposable, file-backed SQLite targets.
+
+Run the same script against each checkout with its src directory on PYTHONPATH.
+Fixture creation/export and restore population are outside validation timings.
+"""
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+import json
+import platform
+import sqlite3
+import statistics
+import tempfile
+import time
+from pathlib import Path
+
+from sqlalchemy import event, insert
+
+from powercontext.builtin.persistence.sources import SourceRepository
+from powercontext.builtin.persistence.sqlite import SQLiteConfig, SQLiteProfile
+from powercontext.builtin.persistence.tables import (
+    BUILTIN_TABLES,
+    SCOPES_TABLE,
+    SOURCE_JOURNAL_HEADS_TABLE,
+    SOURCES_TABLE,
+)
+from powercontext.builtin.portability import PortableBundleService
+from powercontext.builtin.sources import BUILTIN_SOURCE_REGISTRY, ContentSource
+from powercontext.paths import sqlite_url
+from powercontext.sources import SourceMaterialization
+
+
+async def measure(directory: Path, count: int, runs: int) -> None:
+    archive = directory / "sources.pcb"
+
+    async def authorize(_scopes: tuple[str, ...]) -> None:
+        pass
+
+    async with SQLiteProfile.open(
+        SQLiteConfig(url=sqlite_url(directory / "source.db")), tables=BUILTIN_TABLES
+    ) as source:
+        async with source.database.transaction() as connection:
+            await connection.execute(
+                insert(SCOPES_TABLE).values(
+                    scope_id="benchmark",
+                    title="",
+                    summary="",
+                    scope_id_search="benchmark",
+                    title_search="",
+                    summary_search="",
+                    version=1,
+                )
+            )
+            await connection.execute(insert(SOURCE_JOURNAL_HEADS_TABLE).values(scope_id="benchmark", position=count))
+            for start in range(0, count, 100):
+                rows = []
+                for number in range(start, min(start + 100, count)):
+                    identity = f"source-{number:06}"
+                    payload = (
+                        ContentSource(name=identity, materialization=SourceMaterialization.CAPTURED, content="x" * 256)
+                        .model_dump_json()
+                        .encode()
+                    )
+                    rows.append({
+                        "scope_id": "benchmark",
+                        "source_type": "content",
+                        "source_id": identity,
+                        "payload": payload,
+                        "journal_position": number + 1,
+                    })
+                await connection.execute(insert(SOURCES_TABLE), rows)
+        await PortableBundleService(source.database).export(["benchmark"], archive, authorize=authorize)
+
+    for state in ("empty", "identical", "absent"):
+        async with SQLiteProfile.open(
+            SQLiteConfig(url=sqlite_url(directory / f"{state}.db")),
+            tables=() if state == "absent" else BUILTIN_TABLES,
+        ) as target:
+            service = PortableBundleService(
+                target.database, source_decoder=SourceRepository(BUILTIN_SOURCE_REGISTRY).decode_source
+            )
+            restore_seconds = None
+            if state == "identical":
+                started = time.perf_counter()
+                await service.restore(archive)
+                restore_seconds = time.perf_counter() - started
+            executions = []
+
+            def observe(_connection, _cursor, statement, parameters, _context, _many, executions=executions):
+                if statement.startswith("SELECT") and any(
+                    f"FROM {table.name}" in statement
+                    for table in (SCOPES_TABLE, SOURCE_JOURNAL_HEADS_TABLE, SOURCES_TABLE)
+                ):
+                    executions.append(len(parameters))
+
+            event.listen(target.database.engine.sync_engine, "before_cursor_execute", observe)
+            durations, queries = [], []
+            max_parameters = 0
+            for _ in range(runs):
+                executions.clear()
+                started = time.perf_counter()
+                result = await service.validate(archive)
+                durations.append(round(time.perf_counter() - started, 4))
+                if (
+                    not result.compatible
+                    or result.conflicts
+                    or result.already_present != (count + 2 if state == "identical" else 0)
+                ):
+                    message = "unexpected validation classification"
+                    raise RuntimeError(message)
+                queries.append(len(executions))
+                max_parameters = max(max_parameters, max(executions, default=0))
+            print(
+                json.dumps({
+                    "records": count + 2,
+                    "state": state,
+                    "queries": queries,
+                    "seconds": durations,
+                    "median_seconds": statistics.median(durations),
+                    "max_parameters": max_parameters,
+                    "restore_without_projections_seconds": restore_seconds,
+                }),
+                flush=True,
+            )
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--sources", type=int, nargs="+", default=[100, 1_000, 5_000])
+    parser.add_argument("--runs", type=int, default=3)
+    args = parser.parse_args()
+    if args.runs < 1 or any(count < 1 for count in args.sources):
+        parser.error("source counts and runs must be positive")
+    print(
+        json.dumps({
+            "python": platform.python_version(),
+            "sqlite": sqlite3.sqlite_version,
+            "platform": platform.platform(),
+        }),
+        flush=True,
+    )
+    for count in args.sources:
+        with tempfile.TemporaryDirectory(prefix="pc-archive-benchmark-") as temporary:
+            asyncio.run(measure(Path(temporary), count, args.runs))
+
+
+if __name__ == "__main__":
+    main()