fix(migrations): parse arithmetic CHECK constraints

.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 \

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.

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

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
@@ -225,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.")
@@ -526,10 +534,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 +657,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 +714,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 +732,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 +750,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.")