From 579b67e2edfa56df103b5f141b7ad7511cf60f6e Mon Sep 17 00:00:00 2001 From: Frey <370805886@qq.com> Date: Wed, 21 Oct 2026 21:26:06 +0800 Subject: [PATCH 1/2] fix(migrations): parse arithmetic CHECK constraints --- docs/en/docs/operate/database-migrations.md | 6 + docs/zh/docs/operate/database-migrations.md | 2 + .../persistence/migrations/mysql_schema.py | 69 +++++-- .../test_mysql_migration_checks.py | 170 ++++++++++++++++++ tests/fixtures/database_migrations/README.md | 24 +++ .../schemas/seekdb_v1_3_0_checks.json | 25 +++ 6 files changed, 282 insertions(+), 14 deletions(-) create mode 100644 tests/builtin/persistence/test_mysql_migration_checks.py create mode 100644 tests/fixtures/database_migrations/schemas/seekdb_v1_3_0_checks.json diff --git a/docs/en/docs/operate/database-migrations.md b/docs/en/docs/operate/database-migrations.md index dc4239a61b..5e40364468 100644 --- a/docs/en/docs/operate/database-migrations.md +++ b/docs/en/docs/operate/database-migrations.md @@ -27,6 +27,12 @@ through an ODP endpoint on OceanBase AI Database 4.6.3, covering coordinator exc committed DDL, data preservation, and repeat execution. This does not establish complete Server readiness, automatic detection of unfinished remote DDL after a crash, or full Fork restoration acceptance. +MySQL-compatible schema reflection supports arithmetic CHECK expressions with binary `+`, `-`, `*`, `/`, `%`, +unary signs, and nested grouping, preserving arithmetic, comparison, and boolean precedence. Unsupported CHECK +operators/functions remain rejected; their errors identify the reflected table, named constraint, and unsupported +construct instead of misreporting balanced parentheses. Parsing an expression does not register a historical layout +or bypass migration/data-integrity checks. + The existing `server processing-migrate` remains available. Follow [Migrate Artifact processing state](artifact-processing-migration.md); it does not yet forward to the unified entry point. See [Deploy the Server](deploy-server.md) for service deployment. diff --git a/docs/zh/docs/operate/database-migrations.md b/docs/zh/docs/operate/database-migrations.md index 1aa59a040f..bed350885c 100644 --- a/docs/zh/docs/operate/database-migrations.md +++ b/docs/zh/docs/operate/database-migrations.md @@ -14,6 +14,8 @@ description: 了解数据库迁移命令、备份策略、服务启停及当前 输出中的 `ready` 仅表示当前迁移 bundle 已通过结构和数据验证；`readiness_scope=registered_bundle`、`server_ready=false` 明确区分它与完整 Server、索引、旧任务及集群可用。不要以此替代生产升级或完整业务库初始化。普通业务启动的全面版本门禁和完整历史基线尚未纳管。已通过 OceanBase AI 4.6.3 的 ODP 入口完成隔离四表验收，覆盖固定维护主机互斥、DDL 提交后中断的显式确认恢复、数据保留与重复执行。这不代表完整 Server 升级、崩溃后远端 DDL 自动判定或完整 Fork 恢复已通过验收。 +MySQL 兼容后端的结构反射支持 CHECK 表达式中的二元 `+`、`-`、`*`、`/`、`%`、一元正负号及嵌套括号，并保留算术、比较和布尔运算的优先级。未支持的 CHECK 运算符或函数仍被拒绝，错误会标明反射的表名、约束名及未支持的语法，避免将合法分组误报为括号不匹配。表达式解析成功不代表历史结构已登记，也不会绕过迁移或数据完整性检查。 + 原有 `server processing-migrate` 保留，请按[迁移 Artifact 处理状态](artifact-processing-migration.md)使用。它尚未自动转发到统一入口。服务部署参考[部署 Server](deploy-server.md)。 ## Topic Memory 恢复 bundle diff --git a/src/powercontext/builtin/persistence/migrations/mysql_schema.py b/src/powercontext/builtin/persistence/migrations/mysql_schema.py index 1af7804cc9..3449fb7395 100644 --- a/src/powercontext/builtin/persistence/migrations/mysql_schema.py +++ b/src/powercontext/builtin/persistence/migrations/mysql_schema.py @@ -34,7 +34,9 @@ from .bundle import MigrationBundle from .models import MigrationError, digest -_LEXEME = re.compile(r"\s+|`(?:``|[^`])*`|'(?:''|\\.|[^'\\])*'|[A-Za-z_][A-Za-z_0-9$]*|[0-9]+|>=|<=|<>|!=|[(),.=<>+-]") +_LEXEME = re.compile( + r"\s+|`(?:``|[^`])*`|'(?:''|\\.|[^'\\])*'|[A-Za-z_][A-Za-z_0-9$]*|[0-9]+|>=|<=|<>|!=|[(),.=<>+*/%\^&|~!?:@-]" +) _REVISIONS = ("p0001", "p0002", "p0003", "p0004") _CITATION_TABLES = ("pc_artifacts", "pc_artifact_candidate_versions") _FAMILY_CHECK = "ck_pc_artifact_tags_family" @@ -125,7 +127,11 @@ def _reflect_tables(connection: Connection, database: str, exclude: frozenset[st row = connection.exec_driver_sql(f"SHOW CREATE TABLE {_quote(name)}").first() if row is None or not isinstance(row[1], str): raise MigrationError("incompatible_schema", "A table definition could not be reflected.") - reflected_name, table = _parse_table(row[1], database=database) + try: + reflected_name, table = _parse_table(row[1], database=database) + except MigrationError as error: + detail = str(error).removeprefix(f"{error.code}: ") + raise MigrationError(error.code, f"Table {name!r}: {detail}") from error if reflected_name != name: raise MigrationError("incompatible_schema", "A reflected table identity is inconsistent.") inventory["tables"][name] = table @@ -526,10 +532,17 @@ def _put_named(objects: dict[str, Any], name: str, definition: dict[str, Any]) - def _check_definition(part: list[str], constraint: str | None) -> tuple[str, dict[str, Any]]: - expression, end = _group(part, 1) - if constraint is None or part[end:] not in ([], ["enforced"], ["not", "enforced"]): - raise MigrationError("incompatible_schema", "An unsupported CHECK definition was discovered.") - return constraint, {"expression": _Expression(expression).parse(), "enforced": part[end:] != ["not", "enforced"]} + try: + expression, end = _group(part, 1) + if constraint is None or part[end:] not in ([], ["enforced"], ["not", "enforced"]): + raise MigrationError("incompatible_schema", "An unsupported CHECK definition was discovered.") # noqa: TRY301 + return constraint, { + "expression": _Expression(expression).parse(), + "enforced": part[end:] != ["not", "enforced"], + } + except MigrationError as error: + detail = str(error).removeprefix(f"{error.code}: ") + raise MigrationError(error.code, f"CHECK constraint {constraint!r}: {detail}") from error def _index_definition(part: list[str], constraint: str | None = None) -> tuple[str, dict[str, Any]]: @@ -642,12 +655,37 @@ def __init__(self, tokens: list[str]) -> None: def parse(self) -> list[Any]: value = self._expression(0) if self.position != len(self.tokens): - raise MigrationError("incompatible_schema", "An unsupported CHECK expression was discovered.") + lexeme = self.tokens[self.position] + construct = "string literal" if lexeme.startswith("'") else repr(lexeme) + if ( + lexeme == "(" + and self.position > 0 + and re.fullmatch(r"[a-z_][a-z_0-9$]*", self.tokens[self.position - 1]) + ): + construct = f"function {self.tokens[self.position - 1]!r}" + raise MigrationError("incompatible_schema", f"Unsupported CHECK expression token {construct}.") return value def _expression(self, precedence: int) -> list[Any]: left = self._value() - priorities = {"or": 1, "and": 2, "=": 3, ">": 3, "<": 3, ">=": 3, "<=": 3, "!=": 3, "<>": 3, "is": 3, "in": 3} + priorities = { + "or": 1, + "and": 2, + "=": 3, + ">": 3, + "<": 3, + ">=": 3, + "<=": 3, + "!=": 3, + "<>": 3, + "is": 3, + "in": 3, + "+": 4, + "-": 4, + "*": 5, + "/": 5, + "%": 5, + } while self.position < len(self.tokens): operator = self.tokens[self.position] priority = priorities.get(operator, 0) @@ -674,11 +712,12 @@ def _value(self) -> list[Any]: lexeme = self.tokens[self.position] self.position += 1 if lexeme == "(": - value = self._expression(0) - if self.tokens[self.position : self.position + 1] != [")"]: - raise MigrationError("incompatible_schema", "A CHECK expression has unbalanced parentheses.") - self.position += 1 - return value + # Parse a balanced group independently so unsupported operators are + # not mistaken for a missing closing parenthesis. + group, self.position = _group(self.tokens, self.position - 1) + return _Expression(group).parse() + if lexeme in {"+", "-"}: + return ["unary", lexeme, self._value()] if lexeme == "_utf8mb4": if self.position >= len(self.tokens) or not self.tokens[self.position].startswith("'"): raise MigrationError("incompatible_schema", "A CHECK expression has an unsupported literal introducer.") @@ -691,7 +730,7 @@ def _value(self) -> list[Any]: if lexeme == "null": return ["null"] if not re.fullmatch(r"[a-z_][a-z_0-9$]*", lexeme): - raise MigrationError("incompatible_schema", "An unsupported CHECK value was discovered.") + raise MigrationError("incompatible_schema", f"Unsupported CHECK value token {lexeme!r}.") return ["column", lexeme] @@ -709,6 +748,8 @@ def _expression_sql(expression: list[Any]) -> str: return f"({_expression_sql(expression[1])} IS {'NOT ' if kind == 'is_not_null' else ''}NULL)" if kind == "in": return f"({_expression_sql(expression[1])} IN ({', '.join(_expression_sql(value) for value in expression[2])}))" + if kind == "unary": + return f"({expression[1]}{_expression_sql(expression[2])})" if kind == "binary": return f"({_expression_sql(expression[2])} {expression[1].upper()} {_expression_sql(expression[3])})" raise MigrationError("invalid_bundle", "An unsupported frozen CHECK expression was discovered.") diff --git a/tests/builtin/persistence/test_mysql_migration_checks.py b/tests/builtin/persistence/test_mysql_migration_checks.py new file mode 100644 index 0000000000..080dd7229d --- /dev/null +++ b/tests/builtin/persistence/test_mysql_migration_checks.py @@ -0,0 +1,170 @@ +# 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. + +"""CHECK reflection preserves SQL semantics without expanding migration eligibility.""" + +from __future__ import annotations + +import json +import os +from pathlib import Path + +import pytest +from sqlalchemy import create_engine + +from powercontext.builtin.persistence.migrations.bundle import MigrationBundle +from powercontext.builtin.persistence.migrations.connections import MaintenanceConnections +from powercontext.builtin.persistence.migrations.models import MigrationError +from powercontext.builtin.persistence.migrations.mysql_schema import MySQLSchema, _expression_sql, mysql_inventory +from powercontext.builtin.persistence.seekdb import SeekDBConfig +from tests.builtin.persistence.test_mysql_migration_schema import _inventory, _ReflectedDatabase + +_FIXTURE = Path(__file__).parents[2] / "fixtures/database_migrations/schemas/seekdb_v1_3_0_checks.json" +_RESOURCES = Path(__file__).parents[3] / "src/powercontext/builtin/persistence/migrations/resources" + + +def _check_inventory(expression: str): + return _inventory( + _ReflectedDatabase({ + "check_fixture": "CREATE TABLE check_fixture (a INTEGER, b INTEGER, c INTEGER, " + f"CONSTRAINT ck_arithmetic CHECK ({expression}))" + }) + ) + + +@pytest.mark.parametrize( + ("expression", "values", "expected"), + [ + ("a = (b + c)", (5, 2, 3), 1), + ("a = (b + c)", (6, 2, 3), 0), + ("a + b * c = 14", (2, 3, 4), 1), + ("(a + b) * c = 20", (2, 3, 4), 1), + ("a - b - c = 5", (10, 3, 2), 1), + ("a - (b - c) = 9", (10, 3, 2), 1), + ("a / b / c = 2", (16.0, 4, 2), 1), + ("a % b + c = 3", (10, 4, 1), 1), + ("-a + +b = -(c + 1)", (5, 2, 2), 1), + ("a IS NULL OR a = b + c AND b > 0", (None, -2, 3), 1), + ("a IS NULL OR a = b + c AND b > 0", (1, -2, 3), 0), + ("(a IS NULL OR a = b + c) AND b > 0", (None, -2, 3), 0), + ("a + b IN (3, 4) AND c IS NOT NULL", (1, 2, 0), 1), + ("a = b + c", (None, 2, 3), None), + ], +) +def test_reflected_arithmetic_renders_with_equivalent_semantics(expression, values, expected): + inventory = _check_inventory(expression) + rendered = _expression_sql(inventory["tables"]["check_fixture"]["checks"]["ck_arithmetic"]["expression"]) + # SQLite executes the portable expressions; this does not claim MySQL division semantics. + with create_engine("sqlite://").connect() as connection: + result = connection.exec_driver_sql( + f"SELECT {rendered} FROM (SELECT ? AS a, ? AS b, ? AS c)", # noqa: S608 -- test expressions + values, + ).scalar_one() + assert result == expected + assert _check_inventory(rendered) == inventory + + +def test_historical_seekdb_checks_reflect_without_accepting_an_unknown_layout(): + fixture = json.loads(_FIXTURE.read_text()) + inventory = _inventory( + _ReflectedDatabase({name: value["show_create_table"] for name, value in fixture["tables"].items()}) + ) + compiled = _inventory( + _ReflectedDatabase({name: value["create_table"] for name, value in fixture["tables"].items()}) + ) + for name in fixture["tables"]: + # seekdb promotes inline column CHECKs to table constraints in SHOW CREATE. + # Compare the historical table-level checks and round-trip every reflected check. + for constraint, check in compiled["tables"][name]["checks"].items(): + assert inventory["tables"][name]["checks"][constraint] == check + for check in inventory["tables"][name]["checks"].values(): + roundtrip = _check_inventory(_expression_sql(check["expression"])) + assert roundtrip["tables"]["check_fixture"]["checks"]["ck_arithmetic"] == check + checks = inventory["tables"]["pc_recall_effort_daily"]["checks"] + rendered = _expression_sql(checks["ck_pc_recall_effort_daily_drop_total"]["expression"]) + with create_engine("sqlite://").connect() as connection: + for total, expected in [(5, 1), (6, 0)]: + assert ( + connection.exec_driver_sql( + f"SELECT {rendered} FROM (SELECT ? AS dropped_items, " # noqa: S608 -- frozen expression + "2 AS dropped_below_min_bytes, 3 AS dropped_no_fitting_truncation)", + (total,), + ).scalar_one() + == expected + ) + schema = MySQLSchema(MigrationBundle(_RESOURCES)) + with pytest.raises(MigrationError, match="registered complete layout"): + schema.identify(inventory) + with pytest.raises(MigrationError, match="recorded revision"): + schema.verify(inventory, "p0004") + with pytest.raises(MigrationError, match="controlled next-revision"): + schema.recovery(inventory, "p0003") + + +@pytest.mark.parametrize( + ("expression", "construct"), + [ + ("a = (b ^ c)", "'^'"), + ("a = (b DIV c)", "'div'"), + ("a = abs(b)", "'abs'"), + ("a LIKE 'private literal'", "'like'"), + ("a = 'private literal' (b)", "'('"), + ("a = @b", "'@'"), + ], +) +def test_unsupported_checks_report_table_constraint_and_construct(expression, construct): + with pytest.raises(MigrationError) as caught: + _check_inventory(expression) + message = str(caught.value) + assert caught.value.code == "incompatible_schema" + assert "check_fixture" in message + assert "ck_arithmetic" in message + assert construct in message + assert "Unsupported CHECK" in message + assert "unbalanced" not in message + assert "private literal" not in message + + +@pytest.mark.parametrize("expression", ["a = (b + c", "a = b + c)", "a = (b +)", "a = ()", "a = -"]) +def test_malformed_checks_remain_rejected(expression): + with pytest.raises(MigrationError, match="check_fixture"): + _check_inventory(expression) + + +@pytest.mark.skipif(os.environ.get("POWERCONTEXT_TEST_MIGRATION_SEEKDB") != "1", reason="real seekdb probe not enabled") +def test_real_seekdb_historical_check_reflection_and_rendering(short_tmp_path): + pytest.importorskip("pylibseekdb") + fixture = json.loads(_FIXTURE.read_text()) + + def reflect(connection, _identity, guard): + assert connection is not None + for value in fixture["tables"].values(): + guard() + connection.exec_driver_sql(value["create_table"]) + connection.commit() + inventory = mysql_inventory(connection) + checks = inventory["tables"]["pc_recall_effort_daily"]["checks"] + rendered = _expression_sql(checks["ck_pc_recall_effort_daily_drop_total"]["expression"]) + for total, expected in [(5, 1), (6, 0)]: + assert ( + connection.exec_driver_sql( + f"SELECT {rendered} FROM (SELECT {total} AS dropped_items, " # noqa: S608 -- test constants + "2 AS dropped_below_min_bytes, 3 AS dropped_no_fitting_truncation) AS sample" + ).scalar_one() + == expected + ) + with pytest.raises(MigrationError, match="registered complete layout"): + MySQLSchema(MigrationBundle(_RESOURCES)).identify(inventory) + + MaintenanceConnections(SeekDBConfig(path=short_tmp_path / "seekdb")).run(reflect, writable=True) diff --git a/tests/fixtures/database_migrations/README.md b/tests/fixtures/database_migrations/README.md index a8b4494757..888508ca8e 100644 --- a/tests/fixtures/database_migrations/README.md +++ b/tests/fixtures/database_migrations/README.md @@ -25,6 +25,30 @@ The manifest registers exact business fingerprints, alternate released shapes, r Plans bind the canonical target, actual schema, immutable resource snapshot, configuration digest and maintenance choices. Alembic executes the reviewed snapshot even if installed resources change; changes detected during backup stop before migration DDL. +## Historical CHECK reflection fixture + +`schemas/seekdb_v1_3_0_checks.json` records actual `SHOW CREATE TABLE` output for +`pc_recall_effort_daily`, `pc_recall_token_daily`, and `pc_model_usage_daily`. The input DDL was compiled +from `powercontext-v1.3.0` (`95bfbc9e0b59f58aee119c750c69a448b7666d0c`), using that tag's +`tables.py` and `limits.py`, then executed in a fresh disposable embedded seekdb database on macOS. +The fixture records both source blob IDs, the compiler/binding versions, and the engine's `SELECT VERSION()` result. +It was captured with `pylibseekdb 1.4.0.post1` on engine `seekdb-v1.4.0.0`; it is not a copy of a customer's database +or a claim that the original 1.3.0 deployment used this engine version. + +To recapture, load both historical Python sources from the tag, compile each selected table with SQLAlchemy's +MySQL `CreateTable` (and its named indexes with `CreateIndex`), execute them in a new embedded instance, and save +`SHOW CREATE TABLE` without rewriting its expressions. In particular, retain seekdb's nested parentheses, +backticks, literal introducers, and promotion of inline column CHECKs to named table constraints. + +The fixture reproduces #2160 on the original parser. It exercises arithmetic, comparisons, `IN`, and backend formatting; +it is a parser regression input, not a registered migration baseline. Successful reflection must still reject these +unregistered tables during layout identification, verification, and recovery. The opt-in real backend test re-creates +the historical tables and executes the rendered drop-total CHECK against matching and mismatching values: + +```bash +POWERCONTEXT_TEST_MIGRATION_SEEKDB=1 uv run --locked --extra seekdb pytest -q tests/builtin/persistence/test_mysql_migration_checks.py +``` + ## Explicit maintenance Repository-only scratch commands: diff --git a/tests/fixtures/database_migrations/schemas/seekdb_v1_3_0_checks.json b/tests/fixtures/database_migrations/schemas/seekdb_v1_3_0_checks.json new file mode 100644 index 0000000000..b41613f461 --- /dev/null +++ b/tests/fixtures/database_migrations/schemas/seekdb_v1_3_0_checks.json @@ -0,0 +1,25 @@ +{ + "source_tag": "powercontext-v1.3.0", + "source_commit": "95bfbc9e0b59f58aee119c750c69a448b7666d0c", + "source_blobs": { + "src/powercontext/limits.py": "90d8984ea5ed4e35554b7279b7ac8b166d93bc34", + "src/powercontext/builtin/persistence/tables.py": "69c5191fa41c6c7ac9c63ab98e0bb485f539004c" + }, + "pylibseekdb_version": "1.4.0.post1", + "sqlalchemy_version": "2.0.54", + "tables": { + "pc_recall_effort_daily": { + "create_table": "CREATE TABLE pc_recall_effort_daily (\n\tscope_id VARCHAR(256) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL, \n\tusage_date DATE NOT NULL, \n\tpolicy_id VARCHAR(128) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL, \n\tassessment VARCHAR(64) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL, \n\tpreparations BIGINT NOT NULL CONSTRAINT ck_recall_effort_preparations_nonnegative CHECK (preparations >= 0), \n\trounds BIGINT NOT NULL CONSTRAINT ck_recall_effort_rounds_nonnegative CHECK (rounds >= 0), \n\texpanded_preparations BIGINT NOT NULL CONSTRAINT ck_recall_effort_expanded_preparations_nonnegative CHECK (expanded_preparations >= 0), \n\tadmission_expansions BIGINT NOT NULL CONSTRAINT ck_recall_effort_admission_expansions_nonnegative CHECK (admission_expansions >= 0), \n\tpolicy_floor_expansions BIGINT NOT NULL CONSTRAINT ck_recall_effort_policy_floor_expansions_nonnegative CHECK (policy_floor_expansions >= 0), \n\tcandidate_round_samples BIGINT NOT NULL CONSTRAINT ck_recall_effort_candidate_round_samples_nonnegative CHECK (candidate_round_samples >= 0), \n\tcandidates_assessed BIGINT NOT NULL CONSTRAINT ck_recall_effort_candidates_assessed_nonnegative CHECK (candidates_assessed >= 0), \n\tfinal_candidate_pool BIGINT NOT NULL CONSTRAINT ck_recall_effort_final_candidate_pool_nonnegative CHECK (final_candidate_pool >= 0), \n\tadded_embeddings BIGINT NOT NULL CONSTRAINT ck_recall_effort_added_embeddings_nonnegative CHECK (added_embeddings >= 0), \n\tadded_generation_calls BIGINT NOT NULL CONSTRAINT ck_recall_effort_added_generation_calls_nonnegative CHECK (added_generation_calls >= 0), \n\ttruncated_items BIGINT NOT NULL CONSTRAINT ck_recall_effort_truncated_items_nonnegative CHECK (truncated_items >= 0), \n\tdropped_items BIGINT NOT NULL CONSTRAINT ck_recall_effort_dropped_items_nonnegative CHECK (dropped_items >= 0), \n\tdropped_below_min_bytes BIGINT NOT NULL CONSTRAINT ck_recall_effort_dropped_below_min_bytes_nonnegative CHECK (dropped_below_min_bytes >= 0), \n\tdropped_no_fitting_truncation BIGINT NOT NULL CONSTRAINT ck_recall_effort_dropped_no_fitting_truncation_nonnegative CHECK (dropped_no_fitting_truncation >= 0), \n\tPRIMARY KEY (scope_id, usage_date, policy_id, assessment), \n\tCONSTRAINT ck_pc_recall_effort_daily_drop_total CHECK (dropped_items = dropped_below_min_bytes + dropped_no_fitting_truncation)\n)", + "show_create_table": "CREATE TABLE `pc_recall_effort_daily` (\n `scope_id` varchar(256) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL,\n `usage_date` date NOT NULL,\n `policy_id` varchar(128) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL,\n `assessment` varchar(64) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL,\n `preparations` bigint(20) NOT NULL,\n `rounds` bigint(20) NOT NULL,\n `expanded_preparations` bigint(20) NOT NULL,\n `admission_expansions` bigint(20) NOT NULL,\n `policy_floor_expansions` bigint(20) NOT NULL,\n `candidate_round_samples` bigint(20) NOT NULL,\n `candidates_assessed` bigint(20) NOT NULL,\n `final_candidate_pool` bigint(20) NOT NULL,\n `added_embeddings` bigint(20) NOT NULL,\n `added_generation_calls` bigint(20) NOT NULL,\n `truncated_items` bigint(20) NOT NULL,\n `dropped_items` bigint(20) NOT NULL,\n `dropped_below_min_bytes` bigint(20) NOT NULL,\n `dropped_no_fitting_truncation` bigint(20) NOT NULL,\n PRIMARY KEY (`scope_id`, `usage_date`, `policy_id`, `assessment`),\n CONSTRAINT `ck_recall_effort_preparations_nonnegative` CHECK ((`preparations` >= 0)),\n CONSTRAINT `ck_recall_effort_rounds_nonnegative` CHECK ((`rounds` >= 0)),\n CONSTRAINT `ck_recall_effort_expanded_preparations_nonnegative` CHECK ((`expanded_preparations` >= 0)),\n CONSTRAINT `ck_recall_effort_admission_expansions_nonnegative` CHECK ((`admission_expansions` >= 0)),\n CONSTRAINT `ck_recall_effort_policy_floor_expansions_nonnegative` CHECK ((`policy_floor_expansions` >= 0)),\n CONSTRAINT `ck_recall_effort_candidate_round_samples_nonnegative` CHECK ((`candidate_round_samples` >= 0)),\n CONSTRAINT `ck_recall_effort_candidates_assessed_nonnegative` CHECK ((`candidates_assessed` >= 0)),\n CONSTRAINT `ck_recall_effort_final_candidate_pool_nonnegative` CHECK ((`final_candidate_pool` >= 0)),\n CONSTRAINT `ck_recall_effort_added_embeddings_nonnegative` CHECK ((`added_embeddings` >= 0)),\n CONSTRAINT `ck_recall_effort_added_generation_calls_nonnegative` CHECK ((`added_generation_calls` >= 0)),\n CONSTRAINT `ck_recall_effort_truncated_items_nonnegative` CHECK ((`truncated_items` >= 0)),\n CONSTRAINT `ck_recall_effort_dropped_items_nonnegative` CHECK ((`dropped_items` >= 0)),\n CONSTRAINT `ck_recall_effort_dropped_below_min_bytes_nonnegative` CHECK ((`dropped_below_min_bytes` >= 0)),\n CONSTRAINT `ck_recall_effort_dropped_no_fitting_truncation_nonnegative` CHECK ((`dropped_no_fitting_truncation` >= 0)),\n CONSTRAINT `ck_pc_recall_effort_daily_drop_total` CHECK ((`dropped_items` = (`dropped_below_min_bytes` + `dropped_no_fitting_truncation`)))\n) ORGANIZATION INDEX DEFAULT CHARSET = utf8mb4 ROW_FORMAT = DYNAMIC COMPRESSION = 'zstd_1.3.8' BLOCK_SIZE = 16384 TABLET_SIZE = 134217728 PCTFREE = 0" + }, + "pc_recall_token_daily": { + "create_table": "CREATE TABLE pc_recall_token_daily (\n\tscope_id VARCHAR(256) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL, \n\tusage_date DATE NOT NULL, \n\testimator_id VARCHAR(128) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL, \n\testimator_version VARCHAR(64) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL, \n\tpreparations BIGINT NOT NULL, \n\tready_preparations BIGINT NOT NULL, \n\tcomparable_preparations BIGINT NOT NULL, \n\tbaseline_tokens BIGINT NOT NULL, \n\trecalled_tokens BIGINT NOT NULL, \n\tPRIMARY KEY (scope_id, usage_date, estimator_id, estimator_version), \n\tCONSTRAINT ck_pc_recall_token_daily_preparations_nonnegative CHECK (preparations >= 0), \n\tCONSTRAINT ck_pc_recall_token_daily_ready_nonnegative CHECK (ready_preparations >= 0), \n\tCONSTRAINT ck_pc_recall_token_daily_comparable_nonnegative CHECK (comparable_preparations >= 0), \n\tCONSTRAINT ck_pc_recall_token_daily_comparable_ready CHECK (comparable_preparations <= ready_preparations), \n\tCONSTRAINT ck_pc_recall_token_daily_ready_total CHECK (ready_preparations <= preparations), \n\tCONSTRAINT ck_pc_recall_token_daily_baseline_nonnegative CHECK (baseline_tokens >= 0), \n\tCONSTRAINT ck_pc_recall_token_daily_recalled_nonnegative CHECK (recalled_tokens >= 0)\n)", + "show_create_table": "CREATE TABLE `pc_recall_token_daily` (\n `scope_id` varchar(256) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL,\n `usage_date` date NOT NULL,\n `estimator_id` varchar(128) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL,\n `estimator_version` varchar(64) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL,\n `preparations` bigint(20) NOT NULL,\n `ready_preparations` bigint(20) NOT NULL,\n `comparable_preparations` bigint(20) NOT NULL,\n `baseline_tokens` bigint(20) NOT NULL,\n `recalled_tokens` bigint(20) NOT NULL,\n PRIMARY KEY (`scope_id`, `usage_date`, `estimator_id`, `estimator_version`),\n CONSTRAINT `ck_pc_recall_token_daily_preparations_nonnegative` CHECK ((`preparations` >= 0)),\n CONSTRAINT `ck_pc_recall_token_daily_ready_nonnegative` CHECK ((`ready_preparations` >= 0)),\n CONSTRAINT `ck_pc_recall_token_daily_comparable_nonnegative` CHECK ((`comparable_preparations` >= 0)),\n CONSTRAINT `ck_pc_recall_token_daily_comparable_ready` CHECK ((`comparable_preparations` <= `ready_preparations`)),\n CONSTRAINT `ck_pc_recall_token_daily_ready_total` CHECK ((`ready_preparations` <= `preparations`)),\n CONSTRAINT `ck_pc_recall_token_daily_baseline_nonnegative` CHECK ((`baseline_tokens` >= 0)),\n CONSTRAINT `ck_pc_recall_token_daily_recalled_nonnegative` CHECK ((`recalled_tokens` >= 0))\n) ORGANIZATION INDEX DEFAULT CHARSET = utf8mb4 ROW_FORMAT = DYNAMIC COMPRESSION = 'zstd_1.3.8' BLOCK_SIZE = 16384 TABLET_SIZE = 134217728 PCTFREE = 0" + }, + "pc_model_usage_daily": { + "create_table": "CREATE TABLE pc_model_usage_daily (\n\tscope_id VARCHAR(256) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL, \n\tusage_date DATE NOT NULL, \n\tpurpose VARCHAR(64) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL, \n\toperation VARCHAR(16) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL, \n\trequests BIGINT NOT NULL, \n\tinput_tokens BIGINT NOT NULL, \n\toutput_tokens BIGINT NOT NULL, \n\tinput_complete BOOL NOT NULL, \n\toutput_complete BOOL NOT NULL, \n\tPRIMARY KEY (scope_id, usage_date, purpose, operation), \n\tCONSTRAINT ck_pc_model_usage_daily_operation CHECK (operation IN ('generation', 'embedding')), \n\tCONSTRAINT ck_pc_model_usage_daily_requests_nonnegative CHECK (requests >= 0), \n\tCONSTRAINT ck_pc_model_usage_daily_input_nonnegative CHECK (input_tokens >= 0), \n\tCONSTRAINT ck_pc_model_usage_daily_output_nonnegative CHECK (output_tokens >= 0)\n)", + "show_create_table": "CREATE TABLE `pc_model_usage_daily` (\n `scope_id` varchar(256) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL,\n `usage_date` date NOT NULL,\n `purpose` varchar(64) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL,\n `operation` varchar(16) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL,\n `requests` bigint(20) NOT NULL,\n `input_tokens` bigint(20) NOT NULL,\n `output_tokens` bigint(20) NOT NULL,\n `input_complete` tinyint(1) NOT NULL,\n `output_complete` tinyint(1) NOT NULL,\n PRIMARY KEY (`scope_id`, `usage_date`, `purpose`, `operation`),\n CONSTRAINT `ck_pc_model_usage_daily_operation` CHECK ((`operation` in ('generation','embedding'))),\n CONSTRAINT `ck_pc_model_usage_daily_requests_nonnegative` CHECK ((`requests` >= 0)),\n CONSTRAINT `ck_pc_model_usage_daily_input_nonnegative` CHECK ((`input_tokens` >= 0)),\n CONSTRAINT `ck_pc_model_usage_daily_output_nonnegative` CHECK ((`output_tokens` >= 0))\n) ORGANIZATION INDEX DEFAULT CHARSET = utf8mb4 ROW_FORMAT = DYNAMIC COMPRESSION = 'zstd_1.3.8' BLOCK_SIZE = 16384 TABLET_SIZE = 134217728 PCTFREE = 0" + } + }, + "backend_version": "5.7.25-OceanBase seekdb-v1.4.0.0" +} From c4fc8ab7c3e8eb669c23dee6b4d0820583493620 Mon Sep 17 00:00:00 2001 From: Frey <370805886@qq.com> Date: Thu, 22 Oct 2026 10:32:14 +0800 Subject: [PATCH 2/2] fix(migrations): preserve literal percent in CHECK validation --- .github/workflows/master.yml | 1 + .../persistence/migrations/mysql_schema.py | 4 +- .../test_mysql_migration_checks.py | 56 +++++++++++++++++++ tests/fixtures/database_migrations/README.md | 5 ++ 4 files changed, 65 insertions(+), 1 deletion(-) diff --git a/.github/workflows/master.yml b/.github/workflows/master.yml index 01ff65b202..500fa104e2 100644 --- a/.github/workflows/master.yml +++ b/.github/workflows/master.yml @@ -298,6 +298,7 @@ jobs: tests/builtin/persistence/test_database_backup_providers.py \ tests/builtin/persistence/test_database_migration_connections.py \ tests/e2e/test_topic_memory_recovery_migration.py \ + tests/builtin/persistence/test_mysql_migration_checks.py \ tests/builtin/persistence/test_mysql_migration_schema.py -k seekdb \ 2>&1 | tee seekdb-migrations.log uv run --locked --extra seekdb pytest --collect-only -q --color=no \ diff --git a/src/powercontext/builtin/persistence/migrations/mysql_schema.py b/src/powercontext/builtin/persistence/migrations/mysql_schema.py index 3449fb7395..95c5792974 100644 --- a/src/powercontext/builtin/persistence/migrations/mysql_schema.py +++ b/src/powercontext/builtin/persistence/migrations/mysql_schema.py @@ -231,8 +231,10 @@ def validate_data(self, connection: Connection, revision: str) -> None: for name, table in expected["tables"].items(): for check in table["checks"].values(): expression = _expression_sql(check["expression"]) + # No bind parameters: keep literal percent signs out of DBAPI interpolation. violation = connection.exec_driver_sql( - f"SELECT 1 FROM {_quote(name)} WHERE NOT ({expression}) LIMIT 1" # noqa: S608 -- frozen expressions + f"SELECT 1 FROM {_quote(name)} WHERE NOT ({expression}) LIMIT 1", # noqa: S608 -- frozen expressions + execution_options={"no_parameters": True}, ).first() if violation is not None: raise MigrationError("data_integrity_error", "Existing rows violate a registered CHECK constraint.") diff --git a/tests/builtin/persistence/test_mysql_migration_checks.py b/tests/builtin/persistence/test_mysql_migration_checks.py index 080dd7229d..fc779a9d93 100644 --- a/tests/builtin/persistence/test_mysql_migration_checks.py +++ b/tests/builtin/persistence/test_mysql_migration_checks.py @@ -18,6 +18,7 @@ import json import os +import shutil from pathlib import Path import pytest @@ -26,8 +27,10 @@ from powercontext.builtin.persistence.migrations.bundle import MigrationBundle from powercontext.builtin.persistence.migrations.connections import MaintenanceConnections from powercontext.builtin.persistence.migrations.models import MigrationError +from powercontext.builtin.persistence.migrations.mysql import MySQLMigrationRunner from powercontext.builtin.persistence.migrations.mysql_schema import MySQLSchema, _expression_sql, mysql_inventory from powercontext.builtin.persistence.seekdb import SeekDBConfig +from tests.builtin.persistence.test_mysql_migration_runner import _seekdb_scenario_lifetime from tests.builtin.persistence.test_mysql_migration_schema import _inventory, _ReflectedDatabase _FIXTURE = Path(__file__).parents[2] / "fixtures/database_migrations/schemas/seekdb_v1_3_0_checks.json" @@ -168,3 +171,56 @@ def reflect(connection, _identity, guard): MySQLSchema(MigrationBundle(_RESOURCES)).identify(inventory) MaintenanceConnections(SeekDBConfig(path=short_tmp_path / "seekdb")).run(reflect, writable=True) + + +@pytest.mark.skipif(os.environ.get("POWERCONTEXT_TEST_MIGRATION_SEEKDB") != "1", reason="real seekdb probe not enabled") +@pytest.mark.parametrize( + ("expression", "column_type", "valid", "invalid"), + [("a % 2 = 0", "INTEGER", 2, 3), ("a = '100%s%%'", "VARCHAR(32)", "100%s%%", "different")], + ids=["remainder", "percent_literal"], +) +def test_real_seekdb_percent_checks_validate_data_and_plan( + short_tmp_path, monkeypatch, expression, column_type, valid, invalid +): + pytest.importorskip("pylibseekdb") + # Extend an isolated bundle, preserving the shipped layouts and their eligibility. + directory = short_tmp_path / "bundle" + shutil.copytree(_RESOURCES, directory) + ddl = f"CREATE TABLE pc_check_fixture (a {column_type}, CONSTRAINT ck_percent CHECK ({expression}) NOT ENFORCED)" + for name in ("pre_dream", "v1_1_0"): + path = directory / "schemas" / f"{name}.json" + layout = json.loads(path.read_text()) + layout["mysql"].append(ddl) + path.write_text(json.dumps(layout, ensure_ascii=False)) + bundle = MigrationBundle(directory) + schema = MySQLSchema(bundle) + connections = MaintenanceConnections(SeekDBConfig(path=short_tmp_path / "seekdb")) + runner = MySQLMigrationRunner(connections, bundle, evidence_directory=short_tmp_path / "evidence") + + def prepare(connection, _identity, guard): + assert connection is not None + baseline = json.loads((directory / "schemas/pre_dream.json").read_text())["mysql"] + for statement in baseline: + guard() + connection.exec_driver_sql(statement, execution_options={"no_parameters": True}) + connection.exec_driver_sql("INSERT INTO pc_check_fixture (a) VALUES (%s), (NULL)", (valid,)) + connection.commit() + inventory = mysql_inventory(connection) + assert schema.identify(inventory) == "p0001" + schema.verify(inventory, "p0001") + schema.validate_data(connection, "p0001") + + def invalidate(connection, _identity, guard): + assert connection is not None + guard() + connection.exec_driver_sql("UPDATE pc_check_fixture SET a = %s WHERE a IS NOT NULL", (invalid,)) + connection.commit() + with pytest.raises(MigrationError, match=r"data_integrity_error.*CHECK"): + schema.validate_data(connection, "p0001") + + with _seekdb_scenario_lifetime(monkeypatch): + connections.run(prepare, writable=True) + assert runner.plan().state == "migration_required" + connections.run(invalidate, writable=True) + with pytest.raises(MigrationError, match=r"data_integrity_error.*CHECK"): + runner.plan() diff --git a/tests/fixtures/database_migrations/README.md b/tests/fixtures/database_migrations/README.md index 888508ca8e..07ebef7631 100644 --- a/tests/fixtures/database_migrations/README.md +++ b/tests/fixtures/database_migrations/README.md @@ -49,6 +49,11 @@ the historical tables and executes the rendered drop-total CHECK against matchin POWERCONTEXT_TEST_MIGRATION_SEEKDB=1 uv run --locked --extra seekdb pytest -q tests/builtin/persistence/test_mysql_migration_checks.py ``` +The same test module extends a disposable bundle with remainder and percent-literal CHECKs and exercises +`MySQLSchema.validate_data()` and `MySQLMigrationRunner.plan()` through the official aiomysql bridge. +Valid and NULL rows pass; stored violations of the fixture's `NOT ENFORCED` constraints fail with +`data_integrity_error`. These synthetic constraints are not added to the shipped migration layouts. + ## Explicit maintenance Repository-only scratch commands: