From a5ab919937ff145690f092650783d8a790d7a259 Mon Sep 17 00:00:00 2001 From: Mochimia <2755364178@qq.com> Date: Fri, 28 Aug 2026 12:17:48 +0800 Subject: [PATCH 1/3] fix(oceanbase): reject legacy schema collations --- docs/en/docs/how-to/troubleshoot.md | 28 ++++ docs/zh/docs/how-to/troubleshoot.md | 25 +++ .../builtin/persistence/oceanbase/__init__.py | 2 + .../builtin/persistence/oceanbase/profile.py | 59 ++++++++ .../builtin/persistence/tables.py | 7 +- .../persistence/test_oceanbase_profile.py | 142 +++++++++++++++++- 6 files changed, 252 insertions(+), 11 deletions(-) diff --git a/docs/en/docs/how-to/troubleshoot.md b/docs/en/docs/how-to/troubleshoot.md index e52f90c6e..ce85c943b 100644 --- a/docs/en/docs/how-to/troubleshoot.md +++ b/docs/en/docs/how-to/troubleshoot.md @@ -154,6 +154,34 @@ powercontext server run Use the same environment variable whenever you start or diagnose that instance. PowerContext creates missing parent directories for a file-backed SQLite database. +## OceanBase startup rejects an incompatible schema + +Current PowerContext releases compare opaque identity columns byte-for-byte with `utf8mb4_bin`. A database created by +an older release may still use a case-insensitive collation such as `utf8mb4_general_ci`. The Server checks existing +identity columns before creating any missing tables and refuses to start when it finds a mismatch. The startup error +lists each affected `table.column`, its actual collation, and the required collation; it never includes the database +URL or credentials. + +Do not alter these columns in place. They participate in primary keys, foreign keys, and indexes, and an earlier +case-insensitive deployment may already have treated distinct identities as the same value. Use a new empty database +so the previous database remains available for recovery: + +1. Stop the Server and every process that writes to the database. +2. Take and verify a full recoverable backup using your normal OceanBase backup procedure. +3. Export the PowerContext table data with OceanBase `obdumper` in CSV or SQL data mode **without `--ddl`**. Keep the + export and the original database unchanged until the migration is verified. Supply credentials through your + approved secret-handling process rather than placing them in logs or documentation. +4. Create a new empty OceanBase MySQL-mode database and point `POWERCONTEXT_SERVER_DATABASE_URL` at it. Start the + current PowerContext version once to create tables with `utf8mb4_bin`, then stop it before restoring data. +5. Import only the exported row data into the existing new tables with OceanBase `obloader`, again **without + `--ddl`**. Importing the old DDL would recreate the incompatible collations. +6. Compare row counts, inspect the identity-column collations, and test identities that differ only by case or accent. + Start normal traffic only after these checks pass. Retain the old database and backup until the new deployment has + completed your rollback window. + +If records were previously merged because the old collation considered their identities equal, changing the schema +cannot reconstruct them. Resolve those records from an authoritative source before accepting writes. + ## An inference readiness check fails When generation or embedding is configured, Server readiness makes one minimal real provider request. This catches diff --git a/docs/zh/docs/how-to/troubleshoot.md b/docs/zh/docs/how-to/troubleshoot.md index 30d6529c5..26deaa5c0 100644 --- a/docs/zh/docs/how-to/troubleshoot.md +++ b/docs/zh/docs/how-to/troubleshoot.md @@ -150,6 +150,31 @@ powercontext server run 每次启动或诊断该实例时都应使用同一个环境变量。对于文件型 SQLite 数据库，PowerContext 会创建缺失的父 目录。 +## OceanBase 因 schema 不兼容而拒绝启动 + +当前 PowerContext 使用 `utf8mb4_bin` 对不透明 identity column 进行逐字节比较。旧版本创建的数据库可能仍然 +使用 `utf8mb4_general_ci` 等不区分大小写的 collation。Server 会在创建任何缺失表之前检查已有 identity +column；发现不兼容时拒绝启动。启动错误会列出每个受影响的 `table.column`、实际 collation 和要求的 +collation，但不会包含数据库 URL 或凭据。 + +不要直接修改这些 column。它们参与主键、外键和索引，而且旧部署可能已经把本应不同的 identity 当作同一个值。 +请使用新的空数据库，使旧数据库可以继续用于恢复： + +1. 停止 Server 以及所有会写入该数据库的进程。 +2. 按现有 OceanBase 备份流程创建并验证一份可恢复的完整备份。 +3. 使用 OceanBase `obdumper` 的 CSV 或 SQL 数据模式导出 PowerContext 表数据，且**不要使用 `--ddl`**。 + 在迁移验证完成之前，保持导出文件和原数据库不变。请通过获批的 secret 管理流程提供凭据，不要把凭据写入日志 + 或文档。 +4. 新建一个空的 OceanBase MySQL-mode 数据库，将 `POWERCONTEXT_SERVER_DATABASE_URL` 指向它。启动一次当前 + PowerContext，使其创建使用 `utf8mb4_bin` 的表；恢复数据前再次停止 Server。 +5. 使用 OceanBase `obloader` 只把导出的数据导入已经存在的新表，同样**不要使用 `--ddl`**。导入旧 DDL 会重新 + 创建不兼容的 collation。 +6. 比较记录数、检查 identity column collation，并测试仅大小写或重音不同的 identity。全部检查通过后再恢复正常 + 流量。在回滚窗口结束之前，保留旧数据库和备份。 + +如果旧 collation 曾因 identity 相等而合并记录，重建 schema 无法恢复这些记录。接受新的写入之前，请从权威数据源 +修复它们。 + ## 推理服务 readiness 检查失败 配置 generation 或 embedding 后，Server readiness 会向 provider 发起一次最小化真实请求。这样可以发现只有 diff --git a/src/powercontext/builtin/persistence/oceanbase/__init__.py b/src/powercontext/builtin/persistence/oceanbase/__init__.py index 525dce391..8c13de8cd 100644 --- a/src/powercontext/builtin/persistence/oceanbase/__init__.py +++ b/src/powercontext/builtin/persistence/oceanbase/__init__.py @@ -15,12 +15,14 @@ """OceanBase async relational profile.""" from powercontext.builtin.persistence.oceanbase.profile import ( + IncompatibleOceanBaseSchemaError, OceanBaseConfig, OceanBaseProfile, UnsupportedOceanBaseTenantError, ) __all__ = ( + "IncompatibleOceanBaseSchemaError", "OceanBaseConfig", "OceanBaseProfile", "UnsupportedOceanBaseTenantError", diff --git a/src/powercontext/builtin/persistence/oceanbase/profile.py b/src/powercontext/builtin/persistence/oceanbase/profile.py index d6f872b0b..33e35329e 100644 --- a/src/powercontext/builtin/persistence/oceanbase/profile.py +++ b/src/powercontext/builtin/persistence/oceanbase/profile.py @@ -18,6 +18,7 @@ from collections.abc import AsyncIterator from contextlib import asynccontextmanager +from dataclasses import dataclass from typing import Annotated, Literal from pydantic import BaseModel, ConfigDict, Field, SecretStr, field_validator @@ -30,11 +31,36 @@ from powercontext.builtin.persistence.database import AsyncDatabase from powercontext.builtin.persistence.errors import PersistenceError from powercontext.builtin.persistence.schema import create_tables +from powercontext.builtin.persistence.tables import MYSQL_IDENTITY_COLLATION _DIALECT_DRIVER = "mysql+aoceanbase" _DIALECT_REGISTRY_NAME = "mysql.aoceanbase" _DIALECT_MODULE = "pyobvector" _DIALECT_CLASS = "AsyncOceanBaseDialect" +_SCHEMA_COLUMNS_QUERY = """ +SELECT TABLE_NAME, COLUMN_NAME, COLLATION_NAME +FROM information_schema.COLUMNS +WHERE TABLE_SCHEMA = DATABASE() + AND LEFT(TABLE_NAME, 3) = 'pc_' + AND DATA_TYPE = 'varchar' +""" +_SCHEMA_RECREATION_GUIDE = ( + "https://oceanbase.github.io/powercontext/en/docs/how-to/troubleshoot/" + "#oceanbase-startup-rejects-an-incompatible-schema" +) + + +@dataclass(frozen=True) +class _IdentityCollationMismatch: + table_name: str + column_name: str + actual: str | None + + @property + def qualified_name(self) -> str: + """Return the operator-facing table and column name.""" + + return f"{self.table_name}.{self.column_name}" class UnsupportedOceanBaseTenantError(PersistenceError): @@ -46,6 +72,22 @@ def __init__(self, compatibility_mode: str | None) -> None: super().__init__(f"OceanBase profile requires a MySQL-compatible tenant; found {description}") +class IncompatibleOceanBaseSchemaError(PersistenceError): + """Raised when an existing identity column has unsafe comparison semantics.""" + + def __init__(self, mismatches: tuple[_IdentityCollationMismatch, ...]) -> None: + self.columns = tuple(mismatch.qualified_name for mismatch in mismatches) + details = "; ".join( + f"{mismatch.qualified_name} uses {mismatch.actual or 'NULL'} (expected {MYSQL_IDENTITY_COLLATION})" + for mismatch in mismatches + ) + super().__init__( + "OceanBase schema has incompatible identity column collations: " + f"{details}. Back up the database, recreate the PowerContext schema, and restore the data before " + f"restarting. See {_SCHEMA_RECREATION_GUIDE}" + ) + + class OceanBaseConfig(BaseModel): """Validated component configuration for an OceanBase async engine.""" @@ -114,6 +156,7 @@ async def _initialized_profile(profile: OceanBaseProfile) -> AsyncIterator[Ocean try: async with profile.database.transaction() as connection: await _require_mysql_tenant(connection) + await _require_compatible_identity_collations(connection) await create_tables(connection, profile.tables) yield profile finally: @@ -130,6 +173,22 @@ async def _require_mysql_tenant(connection: AsyncConnection) -> None: raise UnsupportedOceanBaseTenantError(mode) +async def _require_compatible_identity_collations(connection: AsyncConnection) -> None: + """Reject legacy PowerContext VARCHAR columns with non-binary identity semantics.""" + + result = await connection.exec_driver_sql(_SCHEMA_COLUMNS_QUERY) + incompatible: list[_IdentityCollationMismatch] = [] + for table_name_value, column_name_value, actual_value in result.all(): + table_name = str(table_name_value) + column_name = str(column_name_value) + actual_collation = None if actual_value is None else str(actual_value) + if actual_collation is None or actual_collation.casefold() != MYSQL_IDENTITY_COLLATION.casefold(): + incompatible.append(_IdentityCollationMismatch(table_name, column_name, actual_collation)) + + if incompatible: + raise IncompatibleOceanBaseSchemaError(tuple(sorted(incompatible, key=lambda item: item.qualified_name))) + + def _register_official_dialect() -> None: # pyobvector publishes the official AsyncOceanBaseDialect but currently # documents explicit SQLAlchemy registry setup instead of an entry point. diff --git a/src/powercontext/builtin/persistence/tables.py b/src/powercontext/builtin/persistence/tables.py index 99e1013a6..c1734f270 100644 --- a/src/powercontext/builtin/persistence/tables.py +++ b/src/powercontext/builtin/persistence/tables.py @@ -46,7 +46,7 @@ SHARED_METADATA = MetaData() -_MYSQL_IDENTITY_COLLATION = "utf8mb4_bin" +MYSQL_IDENTITY_COLLATION = "utf8mb4_bin" def identity_string(length: int): @@ -58,11 +58,12 @@ def identity_string(length: int): ``create_all(checkfirst=True)`` does not rewrite existing column collations, and OceanBase rejects ``ALTER COLUMN ... COLLATE`` when foreign keys exist. - Existing MySQL/OceanBase schemas must be recreated to pick up this type. + The OceanBase profile rejects incompatible existing schemas so operators + can recreate them before the Server accepts work. """ return String(length).with_variant( - VARCHAR(length, charset="utf8mb4", collation=_MYSQL_IDENTITY_COLLATION), + VARCHAR(length, charset="utf8mb4", collation=MYSQL_IDENTITY_COLLATION), "mysql", ) diff --git a/tests/builtin/persistence/test_oceanbase_profile.py b/tests/builtin/persistence/test_oceanbase_profile.py index e7a4d679b..adc6e6c5a 100644 --- a/tests/builtin/persistence/test_oceanbase_profile.py +++ b/tests/builtin/persistence/test_oceanbase_profile.py @@ -27,21 +27,26 @@ from sqlalchemy.ext.asyncio import AsyncEngine, create_async_engine from powercontext.builtin.persistence.oceanbase import ( + IncompatibleOceanBaseSchemaError, OceanBaseConfig, OceanBaseProfile, UnsupportedOceanBaseTenantError, ) from powercontext.builtin.persistence.oceanbase import profile as oceanbase_profile_module +from powercontext.builtin.persistence.tables import SOURCES_TABLE VALID_URL = "mysql+aoceanbase://root%40tenant:secret@127.0.0.1:2881/powercontext?charset=utf8mb4" class _Result: - def __init__(self, row: tuple[str, str] | None = ("ob_compatibility_mode", "MYSQL")) -> None: - self._row = row + def __init__(self, rows: tuple[tuple[object, ...], ...]) -> None: + self._rows = rows + + def first(self) -> tuple[object, ...] | None: + return self._rows[0] if self._rows else None - def first(self) -> tuple[str, str] | None: - return self._row + def all(self) -> list[tuple[object, ...]]: + return list(self._rows) class _Connection: @@ -49,9 +54,23 @@ def __init__(self, row: tuple[str, str] | None = ("ob_compatibility_mode", "MYSQ self.row = row self.statements: list[str] = [] - async def exec_driver_sql(self, statement: str) -> _Result: + async def exec_driver_sql(self, statement: str, parameters: object | None = None) -> _Result: + self.statements.append(statement) + if "ob_compatibility_mode" in statement: + return _Result(() if self.row is None else (self.row,)) + return _Result(()) + + +class _SchemaConnection(_Connection): + def __init__(self, columns: tuple[tuple[object, ...], ...]) -> None: + super().__init__() + self.columns = columns + + async def exec_driver_sql(self, statement: str, parameters: object | None = None) -> _Result: self.statements.append(statement) - return _Result(self.row) + if "ob_compatibility_mode" in statement: + return _Result((("ob_compatibility_mode", "MYSQL"),)) + return _Result(self.columns) class _Begin(AbstractAsyncContextManager[_Connection]): @@ -71,9 +90,14 @@ async def __aexit__( class _Engine: - def __init__(self, *, row: tuple[str, str] | None = ("ob_compatibility_mode", "MYSQL")) -> None: + def __init__( + self, + *, + row: tuple[str, str] | None = ("ob_compatibility_mode", "MYSQL"), + connection: _Connection | None = None, + ) -> None: self.url = make_url(VALID_URL) - self.connection = _Connection(row) + self.connection = connection or _Connection(row) def begin(self) -> _Begin: return _Begin(self.connection) @@ -169,6 +193,71 @@ async def create_no_tables(_connection: object, _tables: tuple[Table, ...]) -> N asyncio.run(scenario()) +def test_profile_rejects_legacy_identity_column_collation(monkeypatch: pytest.MonkeyPatch) -> None: + async def scenario() -> None: + engine = _Engine( + connection=_SchemaConnection(( + ("pc_sources", "source_id", "utf8mb4_unicode_ci"), + ("pc_sources", "scope_id", "utf8mb4_general_ci"), + ("pc_artifacts", "scope_id", "utf8mb4_general_ci"), + )) + ) + created = False + + async def create_no_tables(_connection: object, _tables: tuple[Table, ...]) -> None: + nonlocal created + created = True + + monkeypatch.setattr(oceanbase_profile_module, "create_tables", create_no_tables) + with pytest.raises(IncompatibleOceanBaseSchemaError, match="utf8mb4_general_ci") as caught: + async with OceanBaseProfile.attach(cast(AsyncEngine, engine), tables=()): + pass + + assert not created + assert caught.value.columns == ( + "pc_artifacts.scope_id", + "pc_sources.scope_id", + "pc_sources.source_id", + ) + message = str(caught.value) + assert "pc_sources.scope_id" in message + assert "pc_sources.source_id" in message + assert "pc_artifacts.scope_id" in message + assert "back up" in message.casefold() + assert "recreate" in message.casefold() + assert "restore" in message.casefold() + assert "secret" not in message + + asyncio.run(scenario()) + + +@pytest.mark.parametrize( + "columns", + [ + (), + (("pc_sources", "scope_id", "UTF8MB4_BIN"),), + ], +) +def test_profile_creates_tables_for_empty_or_compatible_schema( + columns: tuple[tuple[object, ...], ...], + monkeypatch: pytest.MonkeyPatch, +) -> None: + async def scenario() -> None: + engine = _Engine(connection=_SchemaConnection(columns)) + created: list[tuple[Table, ...]] = [] + + async def create_selected_tables(_connection: object, tables: tuple[Table, ...]) -> None: + created.append(tables) + + monkeypatch.setattr(oceanbase_profile_module, "create_tables", create_selected_tables) + async with OceanBaseProfile.attach(cast(AsyncEngine, engine), tables=(SOURCES_TABLE,)): + pass + + assert created == [(SOURCES_TABLE,)] + + asyncio.run(scenario()) + + LIVE_URL = os.environ.get("POWERCONTEXT_TEST_OCEANBASE_URL") @@ -186,3 +275,40 @@ async def scenario() -> None: assert await connection.scalar(select(1)) == 1 asyncio.run(scenario()) + + +@pytest.mark.skipif( + not LIVE_URL, + reason="set POWERCONTEXT_TEST_OCEANBASE_URL to a dedicated OceanBase MySQL-mode test database", +) +def test_live_oceanbase_profile_rejects_old_schema_collation() -> None: + async def scenario() -> None: + assert LIVE_URL is not None + config = OceanBaseConfig(url=SecretStr(LIVE_URL)) + async with OceanBaseProfile.open(config, tables=()) as setup_profile: + try: + async with setup_profile.database.transaction() as connection: + await connection.exec_driver_sql("DROP TABLE IF EXISTS `pc_sources`") + await connection.exec_driver_sql( + "CREATE TABLE `pc_sources` (" + "scope_id VARCHAR(256) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NOT NULL, " + "source_type VARCHAR(128) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NOT NULL, " + "source_id VARCHAR(256) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NOT NULL, " + "payload MEDIUMBLOB NOT NULL, journal_position BIGINT NOT NULL, " + "PRIMARY KEY (scope_id, source_type, source_id))" + ) + + with pytest.raises(IncompatibleOceanBaseSchemaError) as caught: + async with OceanBaseProfile.attach(setup_profile.database.engine, tables=(SOURCES_TABLE,)): + pass + + assert caught.value.columns == ( + "pc_sources.scope_id", + "pc_sources.source_id", + "pc_sources.source_type", + ) + finally: + async with setup_profile.database.transaction() as connection: + await connection.exec_driver_sql("DROP TABLE IF EXISTS `pc_sources`") + + asyncio.run(scenario()) From 9f2bfcc42a39c14f5f969c51cdb6d2d2974c4a02 Mon Sep 17 00:00:00 2001 From: Mochimia <2755364178@qq.com> Date: Sat, 29 Aug 2026 11:51:26 +0800 Subject: [PATCH 2/3] fix(oceanbase): isolate identity schema checks --- .../builtin/persistence/oceanbase/profile.py | 19 +++- .../persistence/test_oceanbase_profile.py | 102 +++++++++++++----- 2 files changed, 89 insertions(+), 32 deletions(-) diff --git a/src/powercontext/builtin/persistence/oceanbase/profile.py b/src/powercontext/builtin/persistence/oceanbase/profile.py index 33e35329e..4a1133f98 100644 --- a/src/powercontext/builtin/persistence/oceanbase/profile.py +++ b/src/powercontext/builtin/persistence/oceanbase/profile.py @@ -41,7 +41,6 @@ SELECT TABLE_NAME, COLUMN_NAME, COLLATION_NAME FROM information_schema.COLUMNS WHERE TABLE_SCHEMA = DATABASE() - AND LEFT(TABLE_NAME, 3) = 'pc_' AND DATA_TYPE = 'varchar' """ _SCHEMA_RECREATION_GUIDE = ( @@ -156,7 +155,7 @@ async def _initialized_profile(profile: OceanBaseProfile) -> AsyncIterator[Ocean try: async with profile.database.transaction() as connection: await _require_mysql_tenant(connection) - await _require_compatible_identity_collations(connection) + await _require_compatible_identity_collations(connection, profile.tables) await create_tables(connection, profile.tables) yield profile finally: @@ -173,14 +172,28 @@ async def _require_mysql_tenant(connection: AsyncConnection) -> None: raise UnsupportedOceanBaseTenantError(mode) -async def _require_compatible_identity_collations(connection: AsyncConnection) -> None: +async def _require_compatible_identity_collations( + connection: AsyncConnection, + tables: tuple[Table, ...], +) -> None: """Reject legacy PowerContext VARCHAR columns with non-binary identity semantics.""" + identity_columns = { + (table.name, column.name) + for table in tables + for column in table.columns + if getattr(column.type.dialect_impl(connection.dialect), "collation", None) == MYSQL_IDENTITY_COLLATION + } + if not identity_columns: + return + result = await connection.exec_driver_sql(_SCHEMA_COLUMNS_QUERY) incompatible: list[_IdentityCollationMismatch] = [] for table_name_value, column_name_value, actual_value in result.all(): table_name = str(table_name_value) column_name = str(column_name_value) + if (table_name, column_name) not in identity_columns: + continue actual_collation = None if actual_value is None else str(actual_value) if actual_collation is None or actual_collation.casefold() != MYSQL_IDENTITY_COLLATION.casefold(): incompatible.append(_IdentityCollationMismatch(table_name, column_name, actual_collation)) diff --git a/tests/builtin/persistence/test_oceanbase_profile.py b/tests/builtin/persistence/test_oceanbase_profile.py index adc6e6c5a..dfc3a04f2 100644 --- a/tests/builtin/persistence/test_oceanbase_profile.py +++ b/tests/builtin/persistence/test_oceanbase_profile.py @@ -19,10 +19,12 @@ from contextlib import AbstractAsyncContextManager from types import TracebackType from typing import cast +from uuid import uuid4 import pytest from pydantic import SecretStr, ValidationError -from sqlalchemy import Table, select +from sqlalchemy import Column, MetaData, String, Table, select +from sqlalchemy.dialects.mysql import dialect as mysql_dialect from sqlalchemy.engine import make_url from sqlalchemy.ext.asyncio import AsyncEngine, create_async_engine @@ -33,7 +35,12 @@ UnsupportedOceanBaseTenantError, ) from powercontext.builtin.persistence.oceanbase import profile as oceanbase_profile_module -from powercontext.builtin.persistence.tables import SOURCES_TABLE +from powercontext.builtin.persistence.tables import ( + ARTIFACT_LINEAGE_SOURCES_TABLE, + ARTIFACTS_TABLE, + SOURCES_TABLE, + identity_string, +) VALID_URL = "mysql+aoceanbase://root%40tenant:secret@127.0.0.1:2881/powercontext?charset=utf8mb4" @@ -51,6 +58,7 @@ def all(self) -> list[tuple[object, ...]]: class _Connection: def __init__(self, row: tuple[str, str] | None = ("ob_compatibility_mode", "MYSQL")) -> None: + self.dialect = mysql_dialect() self.row = row self.statements: list[str] = [] @@ -70,6 +78,8 @@ async def exec_driver_sql(self, statement: str, parameters: object | None = None self.statements.append(statement) if "ob_compatibility_mode" in statement: return _Result((("ob_compatibility_mode", "MYSQL"),)) + if "LEFT(TABLE_NAME" in statement: + return _Result(tuple(row for row in self.columns if str(row[0]).startswith("pc_"))) return _Result(self.columns) @@ -195,11 +205,22 @@ async def create_no_tables(_connection: object, _tables: tuple[Table, ...]) -> N def test_profile_rejects_legacy_identity_column_collation(monkeypatch: pytest.MonkeyPatch) -> None: async def scenario() -> None: + extension_table = Table( + "owned_extensions", + MetaData(), + Column("extension_id", identity_string(64)), + Column("display_name", String(64)), + ) engine = _Engine( connection=_SchemaConnection(( + ("owned_extensions", "extension_id", "utf8mb4_general_ci"), + ("owned_extensions", "display_name", "utf8mb4_general_ci"), ("pc_sources", "source_id", "utf8mb4_unicode_ci"), ("pc_sources", "scope_id", "utf8mb4_general_ci"), - ("pc_artifacts", "scope_id", "utf8mb4_general_ci"), + ("pc_artifacts", "family", "utf8mb4_general_ci"), + ("pc_artifact_lineage_sources", "source_type", "utf8mb4_general_ci"), + ("pc_sources", "display_name", "utf8mb4_general_ci"), + ("pc_notes", "display_name", "utf8mb4_general_ci"), )) ) created = False @@ -210,19 +231,27 @@ async def create_no_tables(_connection: object, _tables: tuple[Table, ...]) -> N monkeypatch.setattr(oceanbase_profile_module, "create_tables", create_no_tables) with pytest.raises(IncompatibleOceanBaseSchemaError, match="utf8mb4_general_ci") as caught: - async with OceanBaseProfile.attach(cast(AsyncEngine, engine), tables=()): + async with OceanBaseProfile.attach( + cast(AsyncEngine, engine), + tables=(SOURCES_TABLE, ARTIFACTS_TABLE, ARTIFACT_LINEAGE_SOURCES_TABLE, extension_table), + ): pass assert not created assert caught.value.columns == ( - "pc_artifacts.scope_id", + "owned_extensions.extension_id", + "pc_artifact_lineage_sources.source_type", + "pc_artifacts.family", "pc_sources.scope_id", "pc_sources.source_id", ) message = str(caught.value) + assert "owned_extensions.extension_id" in message + assert "pc_artifact_lineage_sources.source_type" in message + assert "pc_artifacts.family" in message assert "pc_sources.scope_id" in message assert "pc_sources.source_id" in message - assert "pc_artifacts.scope_id" in message + assert "display_name" not in message assert "back up" in message.casefold() assert "recreate" in message.casefold() assert "restore" in message.casefold() @@ -236,6 +265,10 @@ async def create_no_tables(_connection: object, _tables: tuple[Table, ...]) -> N [ (), (("pc_sources", "scope_id", "UTF8MB4_BIN"),), + ( + ("pc_notes", "display_name", "utf8mb4_general_ci"), + ("pc_sources", "display_name", "utf8mb4_general_ci"), + ), ], ) def test_profile_creates_tables_for_empty_or_compatible_schema( @@ -279,36 +312,47 @@ async def scenario() -> None: @pytest.mark.skipif( not LIVE_URL, - reason="set POWERCONTEXT_TEST_OCEANBASE_URL to a dedicated OceanBase MySQL-mode test database", + reason="set POWERCONTEXT_TEST_OCEANBASE_URL to an OceanBase MySQL-mode URL with database creation and deletion privileges", ) def test_live_oceanbase_profile_rejects_old_schema_collation() -> None: async def scenario() -> None: assert LIVE_URL is not None config = OceanBaseConfig(url=SecretStr(LIVE_URL)) - async with OceanBaseProfile.open(config, tables=()) as setup_profile: + database_name = f"pc_test_{uuid4().hex}" + database_created = False + async with OceanBaseProfile.open(config, tables=()) as server_profile: try: - async with setup_profile.database.transaction() as connection: - await connection.exec_driver_sql("DROP TABLE IF EXISTS `pc_sources`") - await connection.exec_driver_sql( - "CREATE TABLE `pc_sources` (" - "scope_id VARCHAR(256) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NOT NULL, " - "source_type VARCHAR(128) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NOT NULL, " - "source_id VARCHAR(256) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NOT NULL, " - "payload MEDIUMBLOB NOT NULL, journal_position BIGINT NOT NULL, " - "PRIMARY KEY (scope_id, source_type, source_id))" + async with server_profile.database.transaction() as connection: + await connection.exec_driver_sql(f"CREATE DATABASE `{database_name}`") + database_created = True + + test_url = make_url(LIVE_URL).set(database=database_name).render_as_string(hide_password=False) + async with OceanBaseProfile.open( + OceanBaseConfig(url=SecretStr(test_url)), + tables=(), + ) as setup_profile: + async with setup_profile.database.transaction() as connection: + await connection.exec_driver_sql( + "CREATE TABLE `pc_sources` (" + "scope_id VARCHAR(256) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NOT NULL, " + "source_type VARCHAR(128) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NOT NULL, " + "source_id VARCHAR(256) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NOT NULL, " + "payload MEDIUMBLOB NOT NULL, journal_position BIGINT NOT NULL, " + "PRIMARY KEY (scope_id, source_type, source_id))" + ) + + with pytest.raises(IncompatibleOceanBaseSchemaError) as caught: + async with OceanBaseProfile.attach(setup_profile.database.engine, tables=(SOURCES_TABLE,)): + pass + + assert caught.value.columns == ( + "pc_sources.scope_id", + "pc_sources.source_id", + "pc_sources.source_type", ) - - with pytest.raises(IncompatibleOceanBaseSchemaError) as caught: - async with OceanBaseProfile.attach(setup_profile.database.engine, tables=(SOURCES_TABLE,)): - pass - - assert caught.value.columns == ( - "pc_sources.scope_id", - "pc_sources.source_id", - "pc_sources.source_type", - ) finally: - async with setup_profile.database.transaction() as connection: - await connection.exec_driver_sql("DROP TABLE IF EXISTS `pc_sources`") + if database_created: + async with server_profile.database.transaction() as connection: + await connection.exec_driver_sql(f"DROP DATABASE `{database_name}`") asyncio.run(scenario()) From 906d6577acb5b21a9c9f7f550cabe04fd55bebdf Mon Sep 17 00:00:00 2001 From: Mochimia <2755364178@qq.com> Date: Sun, 30 Aug 2026 17:28:56 +0800 Subject: [PATCH 3/3] fix(oceanbase): document parent-first data restore --- docs/en/docs/how-to/troubleshoot.md | 48 +++++++++++++-- docs/zh/docs/how-to/troubleshoot.md | 44 ++++++++++++-- .../builtin/persistence/test_mysql_schema.py | 59 ++++++++++++++++++- 3 files changed, 142 insertions(+), 9 deletions(-) diff --git a/docs/en/docs/how-to/troubleshoot.md b/docs/en/docs/how-to/troubleshoot.md index ce85c943b..c15f8e4cd 100644 --- a/docs/en/docs/how-to/troubleshoot.md +++ b/docs/en/docs/how-to/troubleshoot.md @@ -174,10 +174,50 @@ so the previous database remains available for recovery: 4. Create a new empty OceanBase MySQL-mode database and point `POWERCONTEXT_SERVER_DATABASE_URL` at it. Start the current PowerContext version once to create tables with `utf8mb4_bin`, then stop it before restoring data. 5. Import only the exported row data into the existing new tables with OceanBase `obloader`, again **without - `--ddl`**. Importing the old DDL would recreate the incompatible collations. -6. Compare row counts, inspect the identity-column collations, and test identities that differ only by case or accent. - Start normal traffic only after these checks pass. Retain the old database and backup until the new deployment has - completed your rollback window. + `--ddl`**. Keep foreign-key enforcement enabled and run these three layers separately. The examples use CSV; if + you exported SQL data, replace `--csv` with `--sql` in all three commands. Fill in `` through + your approved secret-handling process and make `` select the database created in step 4. + + Before running the commands, compare the exported table files with `SHOW TABLES` in the target database. Every + exported table named below must exist in the target; if one is missing, stop and create it with the current + PowerContext configuration before importing. Remove a name only when the source export does not contain that table. + If the source predates the seven `pc_handoff_report_*` tables, remove them from Layer 1. If the export contains them + but Handoff Report is disabled, temporarily enable it in step 4 to create the current tables, restore their data, + and return the setting to its intended value only after verification. + + Layer 1 contains parents and tables without foreign keys: + + ```bash + obloader -D --csv \ + --table 'pc_source_journal_heads,pc_sources,pc_artifacts,pc_source_cursors,pc_external_skill_registrations,pc_model_usage_daily,pc_recall_token_daily,pc_handoff_report_projects,pc_handoff_report_project_revisions,pc_handoff_report_workstreams,pc_handoff_report_workstream_revisions,pc_handoff_report_workspace_bindings,pc_handoff_report_activity_heads,pc_handoff_report_activities' \ + -f + ``` + + After Layer 1 completes successfully, import its children in Layer 2: + + ```bash + obloader -D --csv \ + --table 'pc_artifact_heads,pc_artifact_lineage_sources,pc_artifact_lineage_artifacts,pc_artifact_candidate_versions,pc_memory_entry_versions' \ + -f + ``` + + After Layer 2 completes successfully, import the remaining children in Layer 3: + + ```bash + obloader -D --csv \ + --table 'pc_artifact_candidate_heads,pc_memory_entry_heads' \ + -f + ``` + + Wait for each invocation to complete successfully before starting the next. Treat any OBLoader error, bad record, + or conflict record as a failed restore. Order within a layer is irrelevant because no table in a layer references + another table in the same layer. +6. If the installation has additional PowerContext-managed tables not listed above, these tested layers do not + classify them. Inspect their foreign-key constraints and place each table after all of its parents; do not add + them to an all-table invocation. +7. Compare source and target row counts for every restored table, inspect the identity-column collations, and test + identities that differ only by case or accent. Start normal traffic only after every check passes. Retain the + source database, verified backup, and export through the rollback window. If records were previously merged because the old collation considered their identities equal, changing the schema cannot reconstruct them. Resolve those records from an authoritative source before accepting writes. diff --git a/docs/zh/docs/how-to/troubleshoot.md b/docs/zh/docs/how-to/troubleshoot.md index 26deaa5c0..1d6dbd835 100644 --- a/docs/zh/docs/how-to/troubleshoot.md +++ b/docs/zh/docs/how-to/troubleshoot.md @@ -167,10 +167,46 @@ collation，但不会包含数据库 URL 或凭据。 或文档。 4. 新建一个空的 OceanBase MySQL-mode 数据库，将 `POWERCONTEXT_SERVER_DATABASE_URL` 指向它。启动一次当前 PowerContext，使其创建使用 `utf8mb4_bin` 的表；恢复数据前再次停止 Server。 -5. 使用 OceanBase `obloader` 只把导出的数据导入已经存在的新表，同样**不要使用 `--ddl`**。导入旧 DDL 会重新 - 创建不兼容的 collation。 -6. 比较记录数、检查 identity column collation，并测试仅大小写或重音不同的 identity。全部检查通过后再恢复正常 - 流量。在回滚窗口结束之前，保留旧数据库和备份。 +5. 使用 OceanBase `obloader` 只把导出的数据导入已经存在的新表，同样**不要使用 `--ddl`**。保持外键检查开启， + 并分别运行下面三层命令。示例使用 CSV；如果导出的是 SQL 数据，请把三个命令中的 `--csv` 全部替换为 + `--sql`。通过获批的 secret 管理流程填写 ``，并让 `` 指向第 4 步创建的 + 数据库。 + + 运行命令前，对照导出的表文件和目标数据库中的 `SHOW TABLES`。下面列出的每张已导出表都必须存在于目标数据库； + 如果目标表缺失，请停止恢复，并先使用当前 PowerContext 配置创建该表。只有源导出不包含某张表时，才能从命令中 + 删除它。源数据库早于七张 `pc_handoff_report_*` 表时，应从第 1 层删除这些表；如果导出包含这些表但 Handoff + Report 已关闭，请在第 4 步临时启用该功能以创建当前表并恢复其数据，验证完成后再恢复预期配置。 + + 第 1 层包含父表和无外键的表： + + ```bash + obloader -D --csv \ + --table 'pc_source_journal_heads,pc_sources,pc_artifacts,pc_source_cursors,pc_external_skill_registrations,pc_model_usage_daily,pc_recall_token_daily,pc_handoff_report_projects,pc_handoff_report_project_revisions,pc_handoff_report_workstreams,pc_handoff_report_workstream_revisions,pc_handoff_report_workspace_bindings,pc_handoff_report_activity_heads,pc_handoff_report_activities' \ + -f + ``` + + 第 1 层成功完成后，导入第 2 层中的子表： + + ```bash + obloader -D --csv \ + --table 'pc_artifact_heads,pc_artifact_lineage_sources,pc_artifact_lineage_artifacts,pc_artifact_candidate_versions,pc_memory_entry_versions' \ + -f + ``` + + 第 2 层成功完成后，导入第 3 层中剩余的子表： + + ```bash + obloader -D --csv \ + --table 'pc_artifact_candidate_heads,pc_memory_entry_heads' \ + -f + ``` + + 每个命令成功完成后才能开始下一层。OBLoader 出现任何错误、bad record 或 conflict record 时，都应判定恢复 + 失败。同一层内的表互不引用，因此层内顺序无关。 +6. 如果安装中还存在上面未列出的 PowerContext 管理表，则这些已测试的层并未对它们分类。检查其外键约束，将每张 + 表放在其所有父表之后；不要把它们加入全表导入命令。 +7. 逐表比较源数据库和目标数据库的记录数，检查 identity column collation，并测试仅大小写或重音不同的 + identity。所有检查通过后才能恢复正常流量。在整个回滚窗口内，保留源数据库、已验证的备份和导出文件。 如果旧 collation 曾因 identity 相等而合并记录，重建 schema 无法恢复这些记录。接受新的写入之前，请从权威数据源 修复它们。 diff --git a/tests/builtin/persistence/test_mysql_schema.py b/tests/builtin/persistence/test_mysql_schema.py index 706ba3df0..3dd435590 100644 --- a/tests/builtin/persistence/test_mysql_schema.py +++ b/tests/builtin/persistence/test_mysql_schema.py @@ -12,12 +12,17 @@ # See the License for the specific language governing permissions and # limitations under the License. -from sqlalchemy import BigInteger, Date, Integer, String +import re +from pathlib import Path + +from sqlalchemy import BigInteger, Date, Integer, String, Table from sqlalchemy.dialects import mysql from sqlalchemy.schema import CreateTable, ForeignKeyConstraint, PrimaryKeyConstraint, UniqueConstraint +from powercontext.builtin.handoff_report.sqlite import HANDOFF_REPORT_TABLES from powercontext.builtin.persistence.tables import ( ARTIFACTS_TABLE, + BUILTIN_TABLES, SHARED_METADATA, SOURCE_CURSORS_TABLE, SOURCES_TABLE, @@ -66,6 +71,58 @@ def test_mysql_ddl_uses_mediumblob_for_every_canonical_payload() -> None: assert f"{column_name} MEDIUMBLOB NOT NULL" in ddl +def _assert_restore_layers_are_parent_first( + restore_layers: tuple[tuple[str, ...], ...], + tables: tuple[Table, ...], +) -> None: + restored_tables = tuple(table_name for layer in restore_layers for table_name in layer) + table_names = {table.name for table in tables} + + assert len(restored_tables) == len(set(restored_tables)) == len(tables) + assert set(restored_tables) == table_names + + foreign_keys = tuple( + (table, constraint) + for table in tables + for constraint in table.constraints + if isinstance(constraint, ForeignKeyConstraint) + ) + + layer_by_table = { + table_name: layer_index for layer_index, layer in enumerate(restore_layers) for table_name in layer + } + for table, foreign_key in foreign_keys: + parent_table_name = foreign_key.referred_table.name + assert layer_by_table[parent_table_name] < layer_by_table[table.name] + + +def test_documented_obloader_restore_layers_are_parent_first() -> None: + restore_guides = ( + Path("docs/en/docs/how-to/troubleshoot.md"), + Path("docs/zh/docs/how-to/troubleshoot.md"), + ) + restore_plans = tuple( + tuple( + tuple(table_names.split(",")) + for table_names in re.findall(r"--table '([^']+)'", guide.read_text(encoding="utf-8")) + ) + for guide in restore_guides + ) + assert all(len(restore_layers) == 3 for restore_layers in restore_plans) + assert len(set(restore_plans)) == 1 + + restore_layers = restore_plans[0] + _assert_restore_layers_are_parent_first(restore_layers, BUILTIN_TABLES + HANDOFF_REPORT_TABLES) + + handoff_report_table_names = {table.name for table in HANDOFF_REPORT_TABLES} + assert handoff_report_table_names <= set(restore_layers[0]) + core_only_layers = tuple( + tuple(table_name for table_name in layer if table_name not in handoff_report_table_names) + for layer in restore_layers + ) + _assert_restore_layers_are_parent_first(core_only_layers, BUILTIN_TABLES) + + def test_every_mysql_utf8mb4_key_stays_below_the_innodb_limit() -> None: budgets: dict[str, int] = {} for table in SHARED_METADATA.tables.values():