diff --git a/docs/en/docs/how-to/troubleshoot.md b/docs/en/docs/how-to/troubleshoot.md index e52f90c6e..c15f8e4cd 100644 --- a/docs/en/docs/how-to/troubleshoot.md +++ b/docs/en/docs/how-to/troubleshoot.md @@ -154,6 +154,74 @@ 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`**. 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. + ## 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..1d6dbd835 100644 --- a/docs/zh/docs/how-to/troubleshoot.md +++ b/docs/zh/docs/how-to/troubleshoot.md @@ -150,6 +150,67 @@ 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`**。保持外键检查开启， + 并分别运行下面三层命令。示例使用 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 无法恢复这些记录。接受新的写入之前，请从权威数据源 +修复它们。 + ## 推理服务 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..4a1133f98 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,35 @@ 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 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 +71,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 +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, profile.tables) await create_tables(connection, profile.tables) yield profile finally: @@ -130,6 +172,36 @@ async def _require_mysql_tenant(connection: AsyncConnection) -> None: raise UnsupportedOceanBaseTenantError(mode) +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)) + + 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_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(): diff --git a/tests/builtin/persistence/test_oceanbase_profile.py b/tests/builtin/persistence/test_oceanbase_profile.py index e7a4d679b..dfc3a04f2 100644 --- a/tests/builtin/persistence/test_oceanbase_profile.py +++ b/tests/builtin/persistence/test_oceanbase_profile.py @@ -19,39 +19,68 @@ 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 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 ( + 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" 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: def __init__(self, row: tuple[str, str] | None = ("ob_compatibility_mode", "MYSQL")) -> None: + self.dialect = mysql_dialect() 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) - return _Result(self.row) + 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) + 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) class _Begin(AbstractAsyncContextManager[_Connection]): @@ -71,9 +100,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 +203,94 @@ 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: + 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", "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 + + 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=(SOURCES_TABLE, ARTIFACTS_TABLE, ARTIFACT_LINEAGE_SOURCES_TABLE, extension_table), + ): + pass + + assert not created + assert caught.value.columns == ( + "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 "display_name" not 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"),), + ( + ("pc_notes", "display_name", "utf8mb4_general_ci"), + ("pc_sources", "display_name", "utf8mb4_general_ci"), + ), + ], +) +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 +308,51 @@ 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 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)) + database_name = f"pc_test_{uuid4().hex}" + database_created = False + async with OceanBaseProfile.open(config, tables=()) as server_profile: + try: + 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", + ) + finally: + if database_created: + async with server_profile.database.transaction() as connection: + await connection.exec_driver_sql(f"DROP DATABASE `{database_name}`") + + asyncio.run(scenario())