diff --git a/backend/app/data/catalog.py b/backend/app/data/catalog.py index 6ca00287..678f249b 100644 --- a/backend/app/data/catalog.py +++ b/backend/app/data/catalog.py @@ -1365,7 +1365,9 @@ def is_date_column(name: str, type_name: str | None) -> bool: 数据剖析（表里只有一个日期类列时提议它作业务日期）和 SQL 检查（按业务日期以外的时间列统计）共用这一个判断。 以前两边各写一套：剖析认的「唯一日期列」，SQL 检查可能认为还有别的时间列，反过来也一样，两边对不上。 """ - t = (type_name or "").upper() + t = _type_words(type_name) + if t.startswith(("ENUM", "SET")) or "INTERVAL" in t: + return False if "DATE" in t or "TIMESTAMP" in t: return True if t and not any(k in t for k in _TEXT_TYPES): @@ -1400,11 +1402,29 @@ def _name_index(tables: Mapping[str, Any]) -> dict[tuple[str, str], set[str]]: return index +#: 类型写法里带引号的部分（ENUM('int','date') 的取值）：认类型时去掉，取值里的词不能让一列变成数字或时间 +_QUOTED_IN_TYPE = re.compile(r"'(?:[^']|'')*'") + + +def _type_words(type_name: str | None) -> str: + """认类型用的写法：大写、去掉引号里的取值。探查存的是按方言编译的完整写法（ENUM('paid','refund')、 + VARCHAR2(20 CHAR)、TIMESTAMP WITH TIME ZONE），不再是 SQLAlchemy 的通用写法。""" + return _QUOTED_IN_TYPE.sub("''", (type_name or "").upper()).strip() + + def _type_family(type_name: str | None) -> str | None: - """列类型的大类：数字、文字、时间；认不出来返回 None（不据此排除）。""" - t = (type_name or "").upper() + """列类型的大类：数字、文字、时间；认不出来返回 None（不据此排除）。 + + 认的是按方言编译的完整写法：MySQL 的 ENUM / SET 是文字（取值列表里的词不算），Oracle 的 INTERVAL DAY TO SECOND + 是时长、不算数字（以前按子串「INT」认成了数字）。PostgreSQL 的枚举编译出来是类型名本身，只看类型认不出来， + 要看列上的取值列表（column_family）。""" + t = _type_words(type_name) if not t: return None + if t.startswith(("ENUM", "SET")): + return "text" + if "INTERVAL" in t: + return None if any(k in t for k in ("CHAR", "TEXT", "CLOB", "STRING", "UUID")): return "text" if any(k in t for k in ("INT", "NUMBER", "NUMERIC", "DECIMAL", "SERIAL", "REAL", "FLOAT", "DOUBLE")): @@ -1414,6 +1434,16 @@ def _type_family(type_name: str | None) -> str | None: return None +def column_family(col: Mapping[str, Any] | None) -> str | None: + """表结构里一列的类型大类：有取值列表（探查记下的 enum_values：MySQL 的 ENUM / SET、PostgreSQL 的枚举）的是 + 文字，其余按类型写法认（_type_family）。""" + if not isinstance(col, Mapping): + return None + if col.get("enum_values"): + return "text" + return _type_family(col.get("type")) + + def _id_stem(column: str) -> str | None: m = _SNAKE_ID.match(column) or _CAMEL_ID.match(column) return m.group("stem") if m else None @@ -1469,8 +1499,8 @@ def infer_name_relations(schema_cache: Mapping[str, Any] | None, table: str, *, pk = list(target_meta.get("primary_key") or []) if len(pk) != 1: continue - pk_type = next((c.get("type") for c in target_meta.get("columns") or [] if c.get("name") == pk[0]), None) - mine, theirs = _type_family(col.get("type")), _type_family(pk_type) + pk_col = next((c for c in target_meta.get("columns") or [] if c.get("name") == pk[0]), None) + mine, theirs = column_family(col), column_family(pk_col) if mine and theirs and mine != theirs: continue out.append(_relation(table, [name], target, pk, "name", cardinality=_cardinality(meta, [name]))) diff --git a/backend/app/data/catalog_profile.py b/backend/app/data/catalog_profile.py index 8eee4488..cc9b1062 100644 --- a/backend/app/data/catalog_profile.py +++ b/backend/app/data/catalog_profile.py @@ -41,7 +41,7 @@ import re import time from collections.abc import Awaitable, Callable from dataclasses import asdict, dataclass, field -from datetime import datetime, timezone +from datetime import date as _date, datetime, timezone from decimal import Decimal from typing import Any @@ -99,7 +99,7 @@ _NUMBER_FIELDS: dict[str, tuple[str, float, float, bool]] = { "table_max_queries": ("每张表最多查询数", 1, 200, True), "table_max_s": ("每张表时长上限（秒）", 10, 600, False), "query_timeout_s": ("单条查询时限（秒）", 1, 60, False), - "sample_size": ("抽样键值数", 10, 10_000, True), + "sample_size": ("核对键值数", 10, 10_000, True), "max_scan_rows": ("整表统计行数上限", 0, 10_000_000, True), } _ENABLED_LABEL = "开启数据剖析" @@ -241,16 +241,82 @@ def sql_literal(value: Any, kind: str | None) -> str | None: return "'" + value.replace("'", "''") + "'" +#: 定长文字类型（CHAR(2)、NCHAR、bpchar、CHARACTER(2)）：PostgreSQL、Oracle 取回的值带尾随空格补齐 +_FIXED_CHAR = re.compile(r"^\s*(?:N?CHAR|BPCHAR|CHARACTER)\b(?!\s+VARYING)", re.IGNORECASE) + + +def is_fixed_char(type_name: str | None) -> bool: + """定长文字列（CHAR(n)）：值的尾随空格是补齐出来的，不是取值的一部分（SQL 比较时也不算）。""" + return bool(type_name) and bool(_FIXED_CHAR.match(str(type_name))) + + +def key_literals(values: list[Any], kind: str | None, *, parent: str | None, + child_type: str | None) -> tuple[list[str], int]: + """从子表取回的键值 → 去父表里对的 IN 列表字面量（去重、保持顺序）：(字面量, 因为两端类型不同没拼的个数)。 + + 字面量按**被指向列**的类型族写（catalog.column_family），不是按子表取回的值： + - 父列是文字、子列是数：写成 '123'。PostgreSQL 上 varchar 列 IN (123) 直接报「operator does not exist」； + - 父列是数、子列是文字：只收整数写法的值（'0012' 也照 12 写），别的（'A1'）计入不同类型、不拼。Oracle 上 + 数字列 IN ('A1') 会把字面量转成数报 ORA-01722，MySQL 会把 'A1' 转成 0 去比、对上不该对上的行； + - 父列是时间：键不核对（各家时间字面量写法不同、精度不同），全部计入不同类型； + - 父列类型认不出：照子表取回的值的类型写（sql_literal，以前的做法）。 + 子表是定长文字（CHAR(n)）时先去掉尾随空格：PG、Oracle 取回的 'A ' 拿去和 VARCHAR 父列比就对不上，而按 CHAR 的 + 语义尾随空格本来就不算。不安全的值（反斜杠、「:名字」、控制字符、太长）照 sql_literal 的规矩跳过，不算类型不同。 + """ + fixed = is_fixed_char(child_type) + out: list[str] = [] + mismatched = 0 + for value in values: + if fixed and isinstance(value, str): + value = value.rstrip(" ") + if parent == "time": + mismatched += 1 + continue + if parent == "text": + if isinstance(value, bool): + continue + if isinstance(value, (int, float, Decimal)): + lit = sql_literal(value, "number") + lit = None if lit is None else "'" + lit + "'" + else: + lit = sql_literal(value, "text") + elif parent == "number": + if isinstance(value, str): + text = value.strip() + if not _INT_TEXT.fullmatch(text): + mismatched += 1 + continue + lit = str(int(text)) + else: + lit = sql_literal(value, "number") + else: + lit = sql_literal(value, kind) + if lit is not None: + out.append(lit) + return list(dict.fromkeys(out)), mismatched + + _DIALECT_MODULES = {"sqlite": sqlite, "postgres": postgresql, "postgresql": postgresql, "mysql": mysql, "mariadb": mysql, "oracle": oracle} +#: 日期列的最小值、最大值只看这个区间里的值（含下限、不含上限）：0000-00-00（MySQL 关了严格模式写进去的零日期）、 +#: 0001-01-01、9999-12-31（「没有」「永久」的占位日期）、PostgreSQL 的 -infinity / infinity 都落在区间外。区间外的 +#: 值照实数出来（「有 N 个无效日期未计入」），不让一个占位日期把业务日期的范围撑成两千年。两端整天排除：带时刻的 +#: 9999-12-31 23:59:59、文字存的「0001-01-01 00:00:00」也算占位 +VALID_DATE_FROM = "0001-01-02" +VALID_DATE_BEFORE = "9999-12-31" +_ISO_DAY = re.compile(r"\d{4}-\d{2}-\d{2}") + + class SqlDialect: - """剖析 SQL 的方言差异：标识符怎么加引号、怎么只取前 n 行、统计信息在哪个系统视图里。只拼字符串，不连库。 + """剖析 SQL 的方言差异：标识符怎么加引号、怎么只取前 n 行、统计信息在哪个系统视图里、日期字面量和按天截断 + 怎么写、有没有块采样。只拼字符串，不连库。 标识符交给 SQLAlchemy 各方言的 identifier_preparer：需要时才加引号（memberTags 加、visits 不加）， MySQL 用反引号。Oracle 尤其不能一律加双引号：探查拿到的表名是 SQLAlchemy 规整过的小写，库里其实是 - 大写，"visits" 加了引号就成了另一张不存在的表；不加引号 Oracle 自己转大写才对得上。 + 大写，"visits" 加了引号就成了另一张不存在的表；不加引号 Oracle 自己转大写才对得上。反过来，库里真是小写的 + 对象（建表时写了 "visits"）探查时记了 quote_name，这种要强制加引号、不转大写（table、column、stats_sql）。 """ def __init__(self, kind: str | None) -> None: @@ -262,13 +328,64 @@ class SqlDialect: self._prep = self._dialect.identifier_preparer self.oracle = self.kind == "oracle" - def quote(self, name: str) -> str: - return self._prep.quote(name) + def quote(self, name: str, *, force: bool = False) -> str: + """标识符。force：探查记了 quote_name 的名字，原样加引号（Oracle 小写带引号的对象）。""" + return self._prep.quote_identifier(name) if force else self._prep.quote(name) def table(self, meta: dict[str, Any] | None, name: str) -> str: """写进 FROM 的表名：探查时指定了 schema 的带上它（Oracle 只读账号名下没有对象，全靠 schema 前缀）。""" schema = (meta or {}).get("schema") - return f"{self._prep.quote_schema(schema)}.{self.quote(name)}" if schema else self.quote(name) + quoted = self.quote(name, force=(meta or {}).get("quote_name") is True) + return f"{self._prep.quote_schema(schema)}.{quoted}" if schema else quoted + + def column(self, meta: dict[str, Any] | None, name: str) -> str: + """写进 SQL 的列名：表结构里这一列记了 quote_name 的强制加引号。""" + cols = (meta or {}).get("columns") or [] + entry = next((c for c in cols if isinstance(c, dict) and c.get("name") == name), None) + return self.quote(name, force=bool(entry and entry.get("quote_name") is True)) + + def date_literal(self, value: Any, *, text: bool = False) -> str: + """日期字面量（取值观察的时间窗口、日期列的有效区间用）。PostgreSQL、MySQL、Oracle 都认 DATE 'YYYY-MM-DD'； + SQLite 没有日期类型，日期存成文字，写字符串。text：列是文字存的日期（VARCHAR），按字符串比——PostgreSQL 上 + varchar 和 date 比较会直接报错。只收真实存在的日期（date 或 YYYY-MM-DD），别的抛 ValueError：拼进 SQL 的 + 只能是我们自己算出来的日子。""" + if isinstance(value, datetime): + value = value.date() + if isinstance(value, _date): + day = value.isoformat() + elif isinstance(value, str) and _ISO_DAY.fullmatch(value): + try: + day = _date.fromisoformat(value).isoformat() + except ValueError: + raise ValueError(f"不是有效的日期：{value}") from None + else: + raise ValueError(f"不是有效的日期：{value!r}") + if text or self.kind == "sqlite": + return f"'{day}'" + return f"DATE '{day}'" + + def day_trunc(self, column: str) -> str: + """时间列截到天（按日行数用）。PostgreSQL CAST(x AS DATE)；MySQL DATE(x)；Oracle 必须用 TRUNC(x)：Oracle 的 + DATE 带时分秒，CAST(ts AS DATE) 截不掉；SQLite 的日期是文字，取前 10 个字符。""" + if self.kind in ("postgres", "postgresql"): + return f"CAST({column} AS DATE)" + if self.kind in ("mysql", "mariadb"): + return f"DATE({column})" + if self.oracle: + return f"TRUNC({column})" + return f"substr({column}, 1, 10)" + + def sample_from(self, table: str, percent: float) -> str | None: + """块采样的 FROM：PostgreSQL TABLESAMPLE SYSTEM (p)、Oracle SAMPLE BLOCK (p)，按数据块随机取约 p% 的行； + MySQL、SQLite 没有块采样，返回 None（调用方退回「按存储顺序的前 N 行」，并照实写口径）。""" + p = float(percent) + if not (0 < p < 100): + raise ValueError(f"采样比例要在 0 到 100 之间：{percent}") + if self.kind in ("postgres", "postgresql"): + return f"{table} TABLESAMPLE SYSTEM ({p:g})" + if self.oracle: + return f"{table} SAMPLE BLOCK ({p:g})" + return None def limit(self, sql: str, n: int) -> str: """只取前 n 行。Oracle 11g 没有 LIMIT 也没有 FETCH FIRST，一律套一层 ROWNUM。""" @@ -304,7 +421,8 @@ class SqlDialect: if self.oracle: denorm = self._dialect.denormalize_name owner = sql_literal(denorm(schema), "text") if schema else "SYS_CONTEXT('USERENV', 'CURRENT_SCHEMA')" - table = sql_literal(denorm(name), "text") + # 小写带引号的对象（quote_name）在数据字典里就是小写，不能转成大写去找 + table = sql_literal(name if (meta or {}).get("quote_name") is True else denorm(name), "text") if not (owner and table): return None return f"SELECT NUM_ROWS FROM ALL_TABLES WHERE OWNER = {owner} AND TABLE_NAME = {table}" @@ -335,8 +453,49 @@ class SqlDialect: f"WHERE {column} IS NOT NULL GROUP BY {column} ORDER BY cnt DESC, {column}") return self.limit(sql, n) - def min_max_sql(self, table: str, columns: list[str]) -> str: - return f"SELECT {', '.join(f'MIN({c}), MAX({c})' for c in columns)} FROM {table}" + def min_max_sql(self, table: str, columns: list[tuple[str, bool]]) -> str: + """日期列的最小值、最大值和无效日期的个数：每列三个值（MIN、MAX、区间外的个数）。columns 是 (写进 SQL 的列, + 是不是文字存的日期)。只看有效区间（VALID_DATE_FROM ≤ x < VALID_DATE_BEFORE）里的值，见 VALID_DATE_FROM。 + 整表统计，只对小表用。""" + parts = [] + for c, text in columns: + lo = self.date_literal(VALID_DATE_FROM, text=text) + hi = self.date_literal(VALID_DATE_BEFORE, text=text) + valid = f"{c} >= {lo} AND {c} < {hi}" + parts.append(f"MIN(CASE WHEN {valid} THEN {c} END), MAX(CASE WHEN {valid} THEN {c} END), " + f"SUM(CASE WHEN {c} < {lo} OR {c} >= {hi} THEN 1 ELSE 0 END)") + return f"SELECT {', '.join(parts)} FROM {table}" + + # ---- 取值观察（下一轮第 3 波用）：窗口、按日行数、块采样。这里先把各方言的写法定下来，金样和容器集成测试都覆盖 + + def _window(self, column: str, since: Any, text: bool) -> str: + """时间窗口的条件：since 起（含），并且早于占位日期 9999-12-31——PostgreSQL 的 infinity、Oracle 的 9999-12-31 + 都大于任何一天，不截掉上端就会在「最近 35 天」里多出一天。""" + return (f"{column} >= {self.date_literal(since, text=text)} " + f"AND {column} < {self.date_literal(VALID_DATE_BEFORE, text=text)}") + + def day_volume_sql(self, table: str, column: str, since: Any, *, text: bool = False) -> str: + """最近若干天按天计数：since 起（含）每天几行。只在日期列打头有索引时对大表发（走索引的范围扫描）。""" + day = self.day_trunc(column) + return (f"SELECT {day} AS d, COUNT(*) AS n FROM {table} " + f"WHERE {self._window(column, since, text)} GROUP BY {day} ORDER BY d") + + def window_value_counts_sql(self, table: str, column: str, date_column: str, since: Any, n: int, *, + text: bool = False) -> str: + """时间窗口里的取值分布：业务日期 since 起（含）的行里，出现最多的 n 个取值和各自的行数。""" + sql = (f"SELECT {column}, COUNT(*) AS cnt FROM {table} " + f"WHERE {self._window(date_column, since, text)} AND {column} IS NOT NULL " + f"GROUP BY {column} ORDER BY cnt DESC, {column}") + return self.limit(sql, n) + + def sample_value_counts_sql(self, table: str, column: str, percent: float, n: int) -> str | None: + """块采样里的取值分布；没有块采样的方言返回 None。""" + source = self.sample_from(table, percent) + if source is None: + return None + sql = (f"SELECT {column}, COUNT(*) AS cnt FROM {source} WHERE {column} IS NOT NULL " + f"GROUP BY {column} ORDER BY cnt DESC, {column}") + return self.limit(sql, n) # ========================================================================== @@ -581,8 +740,10 @@ async def estimate_size(run: Callable[..., Awaitable[QueryResult]], dialect: Sql """先读统计信息；读不到（SQLite、没分析过的表）再数到 max_scan_rows + 1 行为止。 统计信息为 0 或负数（PostgreSQL 没分析过的表是 -1，MySQL 的估算值可能是 0）当作不知道：一张其实很大的表 - 要是被当成空表，后面就会对它做整表统计。返回 (行数, 途中没查成的几条)。这张表的查询次数、时长用完（或者任务止损） -时抛 _Stop。 + 要是被当成空表，后面就会对它做整表统计。统计信息说的行数不超过 max_scan_rows 时也再数一次（多一条有上限的 + 计数）：统计信息可能是很久以前的（表后来涨到了几千万行，ANALYZE 没再跑过），照它把大表当小表，就会对它做 + 整表的 COUNT(DISTINCT)、MIN/MAX。数过的结果是准数（count），数到上限还没数完就按大表算（at_least）。 + 返回 (行数, 途中没查成的几条)。这张表的查询次数、时长用完（或者任务止损）时抛 _Stop。 """ skips: list[_Skip] = [] sql = dialect.stats_sql(meta, name) @@ -592,7 +753,9 @@ async def estimate_size(run: Callable[..., Awaitable[QueryResult]], dialect: Sql except _Skip as e: skips.append(e) else: - if rows is not None and rows > 0: + if rows is not None and rows > max(max_scan_rows, 0): + return TableSize(rows, "stats"), skips + if rows is not None and rows > 0 and max_scan_rows <= 0: return TableSize(rows, "stats"), skips if max_scan_rows <= 0: return TableSize(None, "unknown"), skips @@ -752,6 +915,8 @@ class _CodesOutcome: rows: int note: str complete: bool = False + #: 列定义（ENUM）列出的取值：这次没出现的也写进码值（含义待填），码值是「可能的取值」 + declared: list[str] = field(default_factory=list) @dataclass @@ -901,6 +1066,11 @@ def _column(meta: dict[str, Any], name: str) -> str | None: if isinstance(c, dict) and str(c.get("name", "")).lower() == lowered), None) +def _column_meta(meta: dict[str, Any], name: str) -> dict[str, Any] | None: + """表结构里这一列（类型、取值列表、quote_name）；name 是表结构里的写法（_column 的返回值）。""" + return next((c for c in meta.get("columns") or [] if isinstance(c, dict) and c.get("name") == name), None) + + def _constraint(meta: dict[str, Any], column: str) -> str | None: """这一列单独是主键或某个唯一约束时返回「主键」/「唯一约束」，否则 None。""" wanted = [column.lower()] @@ -950,7 +1120,7 @@ async def _unique(cx: _TableCx, table: str, meta: dict[str, Any], column: str, if not size.within(cx.settings.max_scan_rows): return None, "too_large" d = cx.dialect - res = await cx.run(d.unique_check_sql(d.table(meta, table), d.quote(column))) + res = await cx.run(d.unique_check_sql(d.table(meta, table), d.column(meta, column))) row = res.rows[0] if res.rows else [None, None] total, distinct = _as_int(row[0]), _as_int(row[1]) if total is None or distinct is None: @@ -969,20 +1139,38 @@ def _fk_note(child: tuple[bool | None, str]) -> str: return f"{catalog.PROFILE_NOTE_PREFIX}外键约束已保证引用完整，未抽样核对覆盖率；子表一侧{basis}：{shape}" +#: MySQL 上比较文字按列的排序规则：常见的排序规则不分大小写（_ci），PAD SPACE 的还忽略尾随空格。覆盖率、唯一性、 +#: 码值的分组都按它算，写进说明，人才知道「A01」和「a01」在这里算同一个值 +MYSQL_COLLATION_NOTE = "MySQL 按列的排序规则比较文字（常见的排序规则不分大小写），覆盖率和唯一性按它计算" +MYSQL_CODES_COLLATION_NOTE = "MySQL 按列的排序规则分组（常见的排序规则不分大小写），大小写不同的写法可能并成一个取值" + + +def _collation_note(dialect: SqlDialect, family: str | None, note: str) -> str | None: + """MySQL、MariaDB 上文字列（或类型认不出的列）的排序规则说明；别的方言、数字列没有。""" + return note if dialect.kind in ("mysql", "mariadb") and family in ("text", None) else None + + def _relation_note(*, sample: int, matched: int, coverage: float, scan_cap: int | None, skipped_values: int, parent: tuple[bool | None, str], child: tuple[bool | None, str] | None, - holds: bool) -> str: - """关系的剖析备注：抽样规模、覆盖率、父键和子键唯一的依据。人据此判断这条结论有多可靠。 - 不写日期：什么时候剖析的看这一项的 updated_at（界面按本地时间显示）。""" + holds: bool, mismatched: int = 0, collation: str | None = None) -> str: + """关系的剖析备注：取了多少个键值、看了多少行、覆盖率、父键和子键唯一的依据。人据此判断这条结论有多可靠。 + 不写日期：什么时候剖析的看这一项的 updated_at（界面按本地时间显示）。 + + 不说「抽样」：DISTINCT … LIMIT 取的是数据库先给出的那几个值，大表上是按存储顺序的前若干行里的值，不是随机抽的 + （多半是最老的数据）。mismatched：和被指向列类型不同、没拼进核对的键值个数（key_literals）。""" parts: list[str] = [] low = coverage < catalog.PROFILE_VERIFY_COVERAGE if low: - parts.append(f"抽样覆盖率 {_percent(coverage)}，可能不是这条关系") - scope = f"（只看前 {scan_cap} 行）" if scan_cap else "" - sampled = f"子表抽样 {sample} 个不同键值{scope}，父表对上 {matched} 个" + parts.append(f"覆盖率 {_percent(coverage)}，可能不是这条关系") + scope = f"（按存储顺序只看前 {scan_cap} 行）" if scan_cap else "" + sampled = f"子表取了 {sample} 个不同键值{scope}，父表对上 {matched} 个" parts.append(sampled if low else f"{sampled}，覆盖率 {_percent(coverage)}") if skipped_values: parts.append(f"另有 {skipped_values} 个键值的类型不支持核对，未计入") + if mismatched: + parts.append(f"另有 {mismatched} 个键值和被指向列的类型不同（两端类型不同），未计入") + if collation: + parts.append(collation) unique, how = parent if unique is True: parts.append(f"被指向列是{how}" if how in ("主键", "唯一约束") else "被指向列经计数核对唯一") @@ -1029,18 +1217,25 @@ async def _check_relation(cx: _TableCx, table: str, rel: dict[str, Any], out: _T size = await cx.size_with_data(table, result) scan_cap = cx.scan_cap(size) child_table, parent_table = d.table(meta, table), d.table(pmeta, parent) - res = await cx.run(d.distinct_sample_sql(child_table, d.quote(col), s.sample_size, scan_cap=scan_cap), + res = await cx.run(d.distinct_sample_sql(child_table, d.column(meta, col), s.sample_size, scan_cap=scan_cap), max_rows=s.sample_size) kind = (res.column_types or {}).get(res.columns[0]) if res.columns else None values = [row[0] for row in res.rows if row] - literals = list(dict.fromkeys(lit for v in values if (lit := sql_literal(v, kind)) is not None)) + child_col, parent_col = _column_meta(meta, col), _column_meta(pmeta, pcol) + parent_family = catalog.column_family(parent_col) + literals, mismatched = key_literals(values, kind, parent=parent_family, + child_type=(child_col or {}).get("type")) if not values: raise _Skip("no_data", f"{table}.{col} 没有非空值，无法核对") + if not literals and mismatched: + their = {"number": "数字", "text": "文字", "time": "日期时间"}.get(parent_family or "", "其他类型") + raise _Skip("unsupported", f"两端类型不同：{parent}.{pcol} 是{their}，{table}.{col} 取到的值对不上这种类型，" + "无法核对") if not literals: raise _Skip("unsupported", f"{table}.{col} 的取值类型不支持核对（只核对整数和文字）") matched = 0 for start in range(0, len(literals), IN_CHUNK): - sql = d.match_count_sql(parent_table, d.quote(pcol), literals[start:start + IN_CHUNK]) + sql = d.match_count_sql(parent_table, d.column(pmeta, pcol), literals[start:start + IN_CHUNK]) matched += _as_int(_first(await cx.run(sql))) or 0 matched = min(matched, len(literals)) coverage = round(matched / len(literals), 4) @@ -1051,9 +1246,11 @@ async def _check_relation(cx: _TableCx, table: str, rel: dict[str, Any], out: _T if child_unique is not None: cardinality = "one_to_one" if child_unique[0] is True else "many_to_one" holds = coverage >= catalog.PROFILE_VERIFY_COVERAGE and cardinality is not None + unsafe = len(values) - mismatched - len(literals) note = _relation_note(sample=len(literals), matched=matched, coverage=coverage, scan_cap=scan_cap, - skipped_values=len(values) - len(literals), parent=parent_unique, child=child_unique, - holds=holds) + skipped_values=max(unsafe, 0), parent=parent_unique, child=child_unique, holds=holds, + mismatched=mismatched, + collation=_collation_note(d, parent_family, MYSQL_COLLATION_NOTE)) checked = child_unique is not None and child_unique[0] is not None out.relations.append(_RelationOutcome(rid=str(rel["id"]), target=_relation_target(rel), coverage=coverage, cardinality=cardinality, holds=holds, sample=len(literals), @@ -1076,17 +1273,27 @@ _NOT_CODE_TAILS = frozenset({"id", "no", "num", "number", "name", "desc", "descr "at", "date", "time", "on", "amount", "price", "count", "qty", "rate", "url"}) #: 像日期的取值：2026-07-01、2026/07/01，后面可以跟时间 _DATE_TEXT = re.compile(r"^\d{4}[-/]\d{2}[-/]\d{2}") +#: 日期列最小值、最大值不计入的值（见 VALID_DATE_FROM）在说明里的说法 +INVALID_DATES = "零日期 0000-00-00、0001-01-01、9999-12-31 这类占位日期或无穷大" -def _code_type_ok(type_name: str | None) -> bool: - """整数，或者短文字（定长不超过 MAX_CODE_LEN；SQLite 的 TEXT 不带长度，也算）。""" - t = (type_name or "").upper() +def _is_enum(type_name: str | None, enum_values: Any = None) -> bool: + """取值列表有限的列：MySQL 的 ENUM / SET（类型写法以它们开头），或者探查记了取值列表（PostgreSQL 的枚举）。""" + return bool(enum_values) or catalog._type_words(type_name).startswith(("ENUM", "SET")) + + +def _code_type_ok(type_name: str | None, enum_values: Any = None) -> bool: + """ENUM / SET（取值列表有限，直接算码值候选）、整数，或者短文字（定长不超过 MAX_CODE_LEN；SQLite 的 TEXT 不带 + 长度，也算）。类型是按方言编译的完整写法：VARCHAR2(20 CHAR) 的长度按 20 算。""" + if _is_enum(type_name, enum_values): + return True + t = catalog._type_words(type_name) if not t or any(k in t for k in ("CLOB", "BLOB", "BINARY", "JSON", "DATE", "TIME", "INTERVAL", "REAL", "FLOAT", "DOUBLE", "BOOL")): return False if "INT" in t: return True - size = re.search(r"\(\s*(\d+)\s*(?:,\s*(\d+)\s*)?\)", t) + size = re.search(r"\(\s*(\d+)(?:\s+(?:CHAR|BYTE))?\s*(?:,\s*(\d+)\s*)?\)", t) if any(k in t for k in ("NUMBER", "NUMERIC", "DECIMAL")): return size is None or not size.group(2) or int(size.group(2)) == 0 if any(k in t for k in ("CHAR", "TEXT", "STRING")): @@ -1100,9 +1307,9 @@ def _looks_like_code(name: str) -> bool: def _code_columns(meta: dict[str, Any], notes: dict[str, Any]) -> list[str]: - """要取值分布的列：目录里度量类型记为状态的、剖析以前提过码值候选的（再核一次）、或者类型和命名像状态、 - 类型、渠道的整数或短文字列。主键、唯一约束列、关系里的键列不算；码值人工确认或驳回过的不再取。 - 被遮罩的列也列出来，由检查记为「被遮罩」跳过——要让人看得见为什么它没有码值。""" + """要取值分布的列：目录里度量类型记为状态的、剖析以前提过码值候选的（再核一次）、ENUM / SET 列（取值列表有限， + 不看名字）、或者类型和命名像状态、类型、渠道的整数或短文字列。主键、唯一约束列、关系里的键列不算；码值人工确认 + 或驳回过的不再取。被遮罩的列也列出来，由检查记为「被遮罩」跳过——要让人看得见为什么它没有码值。""" pk = [c.lower() for c in meta.get("primary_key") or []] unique = {str(g[0]).lower() for g in meta.get("unique") or [] if isinstance(g, list) and len(g) == 1} keys = {str(c).lower() for r in notes.get("relations") or [] if isinstance(r, dict) for c in r.get("columns") or []} @@ -1117,23 +1324,28 @@ def _code_columns(meta: dict[str, Any], notes: dict[str, Any]) -> list[str]: status_measure = (isinstance(measure, dict) and measure.get("value") == "status" and measure.get("status") != "rejected") profiled = isinstance(codes, dict) and codes.get("source") == "profile" - if not (status_measure or profiled or _looks_like_code(name)): + enum = _is_enum(col.get("type"), col.get("enum_values")) + if not (status_measure or profiled or enum or _looks_like_code(name)): continue lowered = name.lower() - if (pk == [lowered]) or lowered in unique or lowered in keys or not _code_type_ok(col.get("type")): + if (pk == [lowered]) or lowered in unique or lowered in keys or not _code_type_ok(col.get("type"), + col.get("enum_values")): continue out.append(name) return out -def _code_key(value: Any) -> str | None: - """取值分布里的一个取值 → 码值（文字）。只认整数和不太长的文字。""" +def _code_key(value: Any, *, fixed: bool = False) -> str | None: + """取值分布里的一个取值 → 码值（文字）。只认整数和不太长的文字。fixed：定长文字列（CHAR(n)），去掉补齐的尾随 + 空格（PG、Oracle 取回的是 'A '，WHERE status = 'A' 照样对得上，码值写 'A'）。""" if value is None or isinstance(value, bool): return None if isinstance(value, int): return str(value) if isinstance(value, float): return str(int(value)) if value.is_integer() else None + if isinstance(value, str) and fixed: + value = value.rstrip(" ") if isinstance(value, str) and value.strip() and len(value) <= MAX_CODE_LEN and not _CONTROL.search(value): return value return None @@ -1151,29 +1363,48 @@ async def _check_codes(cx: _TableCx, table: str, column: str, out: _TableOutcome if cx.is_masked(column): raise _Skip("masked", f"{table}.{column} 在数据源设置中被遮罩，不取值", carry=False) d, meta = cx.dialect, cx.tables[table] + col_meta = _column_meta(meta, column) or {} size = await cx.size_with_data(table, result) scan_cap = cx.scan_cap(size) - res = await cx.run(d.value_counts_sql(d.table(meta, table), d.quote(column), CODES_MAX + 1, scan_cap=scan_cap), + res = await cx.run(d.value_counts_sql(d.table(meta, table), d.column(meta, column), CODES_MAX + 1, + scan_cap=scan_cap), max_rows=CODES_MAX + 1) if not res.rows: raise _Skip("no_data", f"{table}.{column} 没有非空值，无法取值分布") if len(res.rows) > CODES_MAX: raise _Skip("high_cardinality", f"{table}.{column} 的取值超过 {CODES_MAX} 个，不作为码值候选", carry=False) - values: list[tuple[str, int]] = [] + fixed = is_fixed_char(col_meta.get("type")) + enum = _is_enum(col_meta.get("type"), col_meta.get("enum_values")) + counts: dict[str, int] = {} for row in res.rows: - key = _code_key(row[0]) + key = _code_key(row[0], fixed=fixed) if key is None: raise _Skip("unsupported", f"{table}.{column} 的取值不是整数或短文字，不作为码值候选", carry=False) - values.append((key, _as_int(row[1]) or 0)) - if len(values) >= 3 and all(n == 1 for _, n in values): - # 每个取值只出现一次（票种编码 T01…T09）：这是一列标识，不是状态码 + counts[key] = counts.get(key, 0) + (_as_int(row[1]) or 0) + values = list(counts.items()) + if len(values) >= 3 and all(n == 1 for _, n in values) and not enum: + # 每个取值只出现一次（票种编码 T01…T09）：这是一列标识，不是状态码。ENUM 的取值列表是列定义给的，不按这条判 raise _Skip("high_cardinality", f"{table}.{column} 的取值各不相同，像标识而不是码值，不作为码值候选", carry=False) rows = sum(n for _, n in values) - scope = f"按前 {scan_cap} 行统计" if scan_cap else "统计全表" + # 不说「抽样」：大表看的是按存储顺序的前若干行（多半是最老的数据），不是随机抽的 + scope = f"按存储顺序取前 {scan_cap} 行统计" if scan_cap else "统计全表" spread = "、".join(f"{k}（{_share(n, rows)}）" for k, n in values) - note = f"{NOTE_PREFIX}{scope}，{rows} 行非空值共 {len(values)} 个取值：{spread}。" - out.codes.append(_CodesOutcome(column=column, values=values, rows=rows, note=note, complete=scan_cap is None)) + extra: list[str] = [] + # ENUM 的取值列表（SET 的是成员，取值是成员的组合，不算）：这次没出现的也是可能的取值 + is_set = catalog._type_words(col_meta.get("type")).startswith("SET") + declared = [] if is_set else [str(v) for v in col_meta.get("enum_values") or []] + unseen = [v for v in declared if v not in counts] + if unseen: + extra.append(f"列定义里另有 {len(unseen)} 个取值这次没有出现：{'、'.join(unseen[:CODES_MAX])}") + family = "text" if enum else catalog.column_family(col_meta) + collation = _collation_note(d, family, MYSQL_CODES_COLLATION_NOTE) + if collation: + extra.append(collation) + tail = "".join(f"；{x}" for x in extra) + note = f"{NOTE_PREFIX}{scope}，{rows} 行非空值共 {len(values)} 个取值：{spread}{tail}。" + out.codes.append(_CodesOutcome(column=column, values=values, rows=rows, note=note, complete=scan_cap is None, + declared=unseen)) def _is_date_column(col: dict[str, Any]) -> bool: @@ -1190,20 +1421,31 @@ async def _check_dates(cx: _TableCx, table: str, columns: list[str], total: int, if not size.within(cx.settings.max_scan_rows): raise _Skip("too_large", f"{table} 的行数超过整表统计的行数上限（{cx.settings.max_scan_rows} 行），" "不取日期列的最小值和最大值") - res = await cx.run(d.min_max_sql(d.table(meta, table), [d.quote(c) for c in columns])) + # 文字存的日期（SQLite 的 TEXT、业务库里的 VARCHAR 日期）按字符串和有效区间比；日期类型的按日期字面量比 + texts = [catalog.column_family(_column_meta(meta, c)) != "time" for c in columns] + res = await cx.run(d.min_max_sql(d.table(meta, table), [(d.column(meta, c), t) for c, t in zip(columns, texts)])) row = res.rows[0] if res.rows else [] + invalid: dict[str, int] = {} for i, column in enumerate(columns): - low, high = (row[2 * i], row[2 * i + 1]) if len(row) >= 2 * i + 2 else (None, None) + low, high, bad = (row[3 * i], row[3 * i + 1], row[3 * i + 2]) if len(row) >= 3 * i + 3 else (None, None, 0) + invalid[column] = _as_int(bad) or 0 if low is None or high is None or not (_DATE_TEXT.match(str(low)) and _DATE_TEXT.match(str(high))): continue - result.date_ranges.append({"column": column, "min": str(low), "max": str(high)}) + found: dict[str, Any] = {"column": column, "min": str(low), "max": str(high)} + if invalid[column]: + found["invalid"] = invalid[column] + result.date_ranges.append(found) if total != 1: return if not result.date_ranges: + if invalid.get(columns[0]): + raise _Skip("no_data", f"{table}.{columns[0]} 只有 {invalid[columns[0]]} 个无效日期（{INVALID_DATES}），" + "没有可识别的日期值") raise _Skip("no_data", f"{table}.{columns[0]} 没有可识别的日期值") found = result.date_ranges[0] + dropped = (f"另有 {found['invalid']} 个无效日期（{INVALID_DATES}）未计入，" if found.get("invalid") else "") note = (f"{NOTE_PREFIX}表里只有这一个日期类列，取值从 {str(found['min'])[:19]} 到 " - f"{str(found['max'])[:19]}，可作为业务日期的参考。") + f"{str(found['max'])[:19]}，{dropped}可作为业务日期的参考。") out.business_date = _DateOutcome(column=found["column"], low=found["min"], high=found["max"], note=note) @@ -1241,6 +1483,8 @@ def _codes_value(c: _CodesOutcome, item: dict[str, Any] | None) -> dict[str, str 沿用它的含义；它写了、数据里没出现的取值也留着——码值是「可能的取值」，没出现不等于不存在。""" old = item.get("value") if isinstance(item, dict) and isinstance(item.get("value"), dict) else {} value = {key: str(old.get(key) or "") for key, _ in c.values} + for key in c.declared: + value.setdefault(key, str(old.get(key) or "")) for key, meaning in old.items(): if key not in value and isinstance(meaning, str) and meaning.strip(): value[key] = meaning @@ -1744,8 +1988,8 @@ def ensure_enabled(row: Any, settings: ProfileSettings | None = None) -> Profile #: 有效行过滤核对的说法（写进这一项的 note；前面是模型给的依据） FILTER_CHECKED = "剖析核对：排除 {share} 的行（共 {rows} 行）" -FILTER_SAMPLED = "剖析核对：按前 {rows} 行统计，排除 {share} 的行" -FILTER_SAMPLED_EDGE = "剖析核对：按前 {rows} 行统计，{what}，请人工核对" +FILTER_SAMPLED = "剖析核对：按存储顺序取前 {rows} 行统计，排除 {share} 的行" +FILTER_SAMPLED_EDGE = "剖析核对：按存储顺序取前 {rows} 行统计，{what}，请人工核对" FILTER_UNCHECKED = "没有核对排除比例（{why}）" FILTER_NOT_ENABLED = "数据源没有开数据剖析" diff --git a/backend/app/data/introspect.py b/backend/app/data/introspect.py index ea8875be..30fbfda2 100644 --- a/backend/app/data/introspect.py +++ b/backend/app/data/introspect.py @@ -165,6 +165,74 @@ def _unique_groups(constraints: list[dict[str, Any]], indexes: list[dict[str, An return groups +def type_text(col_type: Any, dialect: Any) -> str: + """反射到的列类型 → 存进表结构的写法：按这个数据源的方言编译的完整写法（VARCHAR2(20 CHAR)、NUMBER(10, 0)、 + ENUM('paid','refund')、TIMESTAMP WITH TIME ZONE），编译不了的退回 str()。 + + 以前存的是 str(类型)，即 SQLAlchemy 的通用写法：Oracle 的 VARCHAR2、NUMBER 丢了长度和精度，MySQL 的 ENUM 丢了 + 取值列表，PostgreSQL 的 TIMESTAMP 丢了时区。类型写不全，表结构前后比对就看不出类型变了，ENUM 状态列也进不了 + 码值候选。这一改会让新探查的表结构快照哈希变化（提示词里的类型写法变长），旧运行封存的快照不受影响。""" + if col_type is None: + return "" + try: + compiled = str(col_type.compile(dialect=dialect)) + except Exception: # noqa: BLE001 - 编译不了（NullType 等）就用通用写法，不让一列拖垮整次探查 + return str(col_type) + if getattr(dialect, "name", "") == "oracle" and compiled.startswith("VARCHAR("): + # SQLAlchemy 把 Oracle 的 VARCHAR2 反射成通用的 VARCHAR，按 Oracle 方言编译出来是 VARCHAR(10 CHAR)（容器集成 + # 测试实测）。Oracle 的数据字典里只有 VARCHAR2（建表写 VARCHAR 也存成 VARCHAR2），照数据字典写 + compiled = "VARCHAR2" + compiled[len("VARCHAR"):] + return compiled + + +def _enum_values(col_type: Any) -> list[str] | None: + """ENUM（MySQL、PostgreSQL 的枚举，都是 sqltypes.Enum 的子类）的取值列表，MySQL SET 的成员列表；别的类型 None。""" + from sqlalchemy.dialects.mysql import SET + from sqlalchemy.types import Enum + + if isinstance(col_type, Enum): + values = list(col_type.enums or []) + elif isinstance(col_type, SET): + values = list(col_type.values or []) + else: + return None + return [str(v) for v in values] + + +def _forced_quote(name: Any) -> bool: + """名字必须原样加引号：Oracle 反射回来的全小写对象（建表时写了 "visits"）是 quoted_name(quote=True)。 + 它存进 JSON 就成了普通字符串，剖析拼 SQL 时不加引号，Oracle 把它转成大写 VISITS，查的是另一个不存在的对象。""" + return getattr(name, "quote", None) is True + + +def column_entry(col: Mapping[str, Any], dialect: Any) -> dict[str, Any]: + """反射到的一列 → 表结构里的一列：名字、按方言编译的类型、可空、注释；ENUM / SET 另记取值列表（enum_values）， + 要强制加引号的名字另记 quote_name。只在有的时候写这两个键：其余列和以前一字不差。""" + entry: dict[str, Any] = { + "name": str(col["name"]), + "type": type_text(col.get("type"), dialect), + "nullable": bool(col.get("nullable", True)), + "comment": col.get("comment") or None, + } + values = _enum_values(col.get("type")) + if values: + entry["enum_values"] = values + if _forced_quote(col.get("name")): + entry["quote_name"] = True + return entry + + +def index_first_columns(indexes: list[dict[str, Any]], pk: list[str]) -> list[str]: + """哪些列打头有索引：各索引（含主键）的第一列，去重、按名字排序。取值观察看大表的最新业务日期、按日行数时， + 只在业务日期列打头有索引时才发窗口查询（按它走索引，不扫全表）。表达式索引（列名是 None）不算。""" + first = {str(pk[0])} if pk else set() + for ix in indexes: + cols = ix.get("column_names") or [] + if cols and cols[0] is not None: + first.add(str(cols[0])) + return sorted(first) + + async def introspect(source: Any, *, schema: str | None = None) -> dict[str, Any]: """探查整个库的结构，结果直接存进 DataSource.schema_cache。 @@ -266,24 +334,20 @@ async def introspect(source: Any, *, schema: str | None = None) -> dict[str, Any # 只读账号名下没有这些对象，全靠跨 schema 授权 "qualified": f"{target_schema}.{name}" if target_schema else name, "schema": target_schema, - "columns": [ - { - "name": c["name"], - "type": str(c.get("type") or ""), - "nullable": bool(c.get("nullable", True)), - "comment": c.get("comment") or None, - } - for c in columns - ], + "columns": [column_entry(c, sync_conn.dialect) for c in columns], "primary_key": pk, "comment": comment, "is_view": name in views, } + if _forced_quote(name): + tables[name]["quote_name"] = True if name in fk_ok: tables[name]["foreign_keys"] = _foreign_keys(fk_by.get(key) or [], target_schema, default_schema) if name in uq_ok: tables[name]["unique"] = _unique_groups(uq_by.get(key) or [], (ix_by.get(key) or []) if name in ix_ok else [], pk) + if name in ix_ok: + tables[name]["index_first_columns"] = index_first_columns(ix_by.get(key) or [], pk) payload: dict[str, Any] = {"tables": tables, "truncated": truncated, "total": len(names) + len(views)} if failures and not tables: @@ -471,6 +535,10 @@ def describe_table(source: Any, table: str, *, catalog: Mapping[str, Mapping[str marks.append("非空") if col.get("comment"): marks.append(col["comment"]) + values = col.get("enum_values") + if values and "(" not in str(col.get("type") or ""): + # PostgreSQL 的枚举类型写法只有类型名（order_status）：取值列表另外写出来，模型才知道能写哪些值 + marks.append("取值：" + "、".join(str(v) for v in values[:30]) + ("…" if len(values) > 30 else "")) suffix = f" {'、'.join(marks)}" if marks else "" lines.append(f" {col['name']} {col['type']}{suffix}") if catalog: diff --git a/backend/tests/fixtures/dialect/mariadb.sql b/backend/tests/fixtures/dialect/mariadb.sql new file mode 100644 index 00000000..0130c4d9 --- /dev/null +++ b/backend/tests/fixtures/dialect/mariadb.sql @@ -0,0 +1,27 @@ +-- 剖析与取值观察的 SQL 金样：mariadb（tests/test_profile_dialect_golden.py 生成，改动要人审） +-- stats +SELECT TABLE_ROWS FROM information_schema.TABLES WHERE TABLE_SCHEMA = 'sales' AND TABLE_NAME = 'orders'; +-- stats_quoted +SELECT TABLE_ROWS FROM information_schema.TABLES WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = 'visits'; +-- bounded +SELECT COUNT(*) AS n FROM (SELECT 1 AS one FROM sales.orders LIMIT 100001) s; +-- distinct +SELECT DISTINCT store_code FROM sales.orders WHERE store_code IS NOT NULL LIMIT 2000; +-- distinct_capped +SELECT DISTINCT store_code FROM (SELECT store_code FROM sales.orders WHERE store_code IS NOT NULL LIMIT 100000) s LIMIT 2000; +-- match +SELECT COUNT(DISTINCT code) AS n FROM stores WHERE code IN ('12', '7'); +-- unique +SELECT COUNT(`level`) AS n, COUNT(DISTINCT `level`) AS d FROM `visits`; +-- value_counts +SELECT status, COUNT(*) AS cnt FROM sales.orders WHERE status IS NOT NULL GROUP BY status ORDER BY cnt DESC, status LIMIT 21; +-- value_counts_capped +SELECT status, COUNT(*) AS cnt FROM (SELECT status FROM sales.orders WHERE status IS NOT NULL LIMIT 100000) s WHERE status IS NOT NULL GROUP BY status ORDER BY cnt DESC, status LIMIT 21; +-- min_max +SELECT MIN(CASE WHEN paid_at >= DATE '0001-01-02' AND paid_at < DATE '9999-12-31' THEN paid_at END), MAX(CASE WHEN paid_at >= DATE '0001-01-02' AND paid_at < DATE '9999-12-31' THEN paid_at END), SUM(CASE WHEN paid_at < DATE '0001-01-02' OR paid_at >= DATE '9999-12-31' THEN 1 ELSE 0 END), MIN(CASE WHEN biz_day >= '0001-01-02' AND biz_day < '9999-12-31' THEN biz_day END), MAX(CASE WHEN biz_day >= '0001-01-02' AND biz_day < '9999-12-31' THEN biz_day END), SUM(CASE WHEN biz_day < '0001-01-02' OR biz_day >= '9999-12-31' THEN 1 ELSE 0 END) FROM sales.orders; +-- date_window +SELECT status, COUNT(*) AS cnt FROM sales.orders WHERE paid_at >= DATE '2026-08-27' AND paid_at < DATE '9999-12-31' AND status IS NOT NULL GROUP BY status ORDER BY cnt DESC, status LIMIT 21; +-- day_volume +SELECT DATE(paid_at) AS d, COUNT(*) AS n FROM sales.orders WHERE paid_at >= DATE '2026-08-27' AND paid_at < DATE '9999-12-31' GROUP BY DATE(paid_at) ORDER BY d; +-- sample +-- （这种方言没有这类查询） diff --git a/backend/tests/fixtures/dialect/mysql.sql b/backend/tests/fixtures/dialect/mysql.sql new file mode 100644 index 00000000..1b046e36 --- /dev/null +++ b/backend/tests/fixtures/dialect/mysql.sql @@ -0,0 +1,27 @@ +-- 剖析与取值观察的 SQL 金样：mysql（tests/test_profile_dialect_golden.py 生成，改动要人审） +-- stats +SELECT TABLE_ROWS FROM information_schema.TABLES WHERE TABLE_SCHEMA = 'sales' AND TABLE_NAME = 'orders'; +-- stats_quoted +SELECT TABLE_ROWS FROM information_schema.TABLES WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = 'visits'; +-- bounded +SELECT COUNT(*) AS n FROM (SELECT 1 AS one FROM sales.orders LIMIT 100001) s; +-- distinct +SELECT DISTINCT store_code FROM sales.orders WHERE store_code IS NOT NULL LIMIT 2000; +-- distinct_capped +SELECT DISTINCT store_code FROM (SELECT store_code FROM sales.orders WHERE store_code IS NOT NULL LIMIT 100000) s LIMIT 2000; +-- match +SELECT COUNT(DISTINCT code) AS n FROM stores WHERE code IN ('12', '7'); +-- unique +SELECT COUNT(`level`) AS n, COUNT(DISTINCT `level`) AS d FROM `visits`; +-- value_counts +SELECT status, COUNT(*) AS cnt FROM sales.orders WHERE status IS NOT NULL GROUP BY status ORDER BY cnt DESC, status LIMIT 21; +-- value_counts_capped +SELECT status, COUNT(*) AS cnt FROM (SELECT status FROM sales.orders WHERE status IS NOT NULL LIMIT 100000) s WHERE status IS NOT NULL GROUP BY status ORDER BY cnt DESC, status LIMIT 21; +-- min_max +SELECT MIN(CASE WHEN paid_at >= DATE '0001-01-02' AND paid_at < DATE '9999-12-31' THEN paid_at END), MAX(CASE WHEN paid_at >= DATE '0001-01-02' AND paid_at < DATE '9999-12-31' THEN paid_at END), SUM(CASE WHEN paid_at < DATE '0001-01-02' OR paid_at >= DATE '9999-12-31' THEN 1 ELSE 0 END), MIN(CASE WHEN biz_day >= '0001-01-02' AND biz_day < '9999-12-31' THEN biz_day END), MAX(CASE WHEN biz_day >= '0001-01-02' AND biz_day < '9999-12-31' THEN biz_day END), SUM(CASE WHEN biz_day < '0001-01-02' OR biz_day >= '9999-12-31' THEN 1 ELSE 0 END) FROM sales.orders; +-- date_window +SELECT status, COUNT(*) AS cnt FROM sales.orders WHERE paid_at >= DATE '2026-08-27' AND paid_at < DATE '9999-12-31' AND status IS NOT NULL GROUP BY status ORDER BY cnt DESC, status LIMIT 21; +-- day_volume +SELECT DATE(paid_at) AS d, COUNT(*) AS n FROM sales.orders WHERE paid_at >= DATE '2026-08-27' AND paid_at < DATE '9999-12-31' GROUP BY DATE(paid_at) ORDER BY d; +-- sample +-- （这种方言没有这类查询） diff --git a/backend/tests/fixtures/dialect/oracle.sql b/backend/tests/fixtures/dialect/oracle.sql new file mode 100644 index 00000000..c338ead9 --- /dev/null +++ b/backend/tests/fixtures/dialect/oracle.sql @@ -0,0 +1,27 @@ +-- 剖析与取值观察的 SQL 金样：oracle（tests/test_profile_dialect_golden.py 生成，改动要人审） +-- stats +SELECT NUM_ROWS FROM ALL_TABLES WHERE OWNER = 'SALES' AND TABLE_NAME = 'ORDERS'; +-- stats_quoted +SELECT NUM_ROWS FROM ALL_TABLES WHERE OWNER = SYS_CONTEXT('USERENV', 'CURRENT_SCHEMA') AND TABLE_NAME = 'visits'; +-- bounded +SELECT COUNT(*) AS n FROM (SELECT * FROM (SELECT 1 AS one FROM sales.orders) WHERE ROWNUM <= 100001) s; +-- distinct +SELECT * FROM (SELECT DISTINCT store_code FROM sales.orders WHERE store_code IS NOT NULL) WHERE ROWNUM <= 2000; +-- distinct_capped +SELECT * FROM (SELECT DISTINCT store_code FROM (SELECT * FROM (SELECT store_code FROM sales.orders WHERE store_code IS NOT NULL) WHERE ROWNUM <= 100000) s) WHERE ROWNUM <= 2000; +-- match +SELECT COUNT(DISTINCT code) AS n FROM stores WHERE code IN ('12', '7'); +-- unique +SELECT COUNT("level") AS n, COUNT(DISTINCT "level") AS d FROM "visits"; +-- value_counts +SELECT * FROM (SELECT status, COUNT(*) AS cnt FROM sales.orders WHERE status IS NOT NULL GROUP BY status ORDER BY cnt DESC, status) WHERE ROWNUM <= 21; +-- value_counts_capped +SELECT * FROM (SELECT status, COUNT(*) AS cnt FROM (SELECT * FROM (SELECT status FROM sales.orders WHERE status IS NOT NULL) WHERE ROWNUM <= 100000) s WHERE status IS NOT NULL GROUP BY status ORDER BY cnt DESC, status) WHERE ROWNUM <= 21; +-- min_max +SELECT MIN(CASE WHEN paid_at >= DATE '0001-01-02' AND paid_at < DATE '9999-12-31' THEN paid_at END), MAX(CASE WHEN paid_at >= DATE '0001-01-02' AND paid_at < DATE '9999-12-31' THEN paid_at END), SUM(CASE WHEN paid_at < DATE '0001-01-02' OR paid_at >= DATE '9999-12-31' THEN 1 ELSE 0 END), MIN(CASE WHEN biz_day >= '0001-01-02' AND biz_day < '9999-12-31' THEN biz_day END), MAX(CASE WHEN biz_day >= '0001-01-02' AND biz_day < '9999-12-31' THEN biz_day END), SUM(CASE WHEN biz_day < '0001-01-02' OR biz_day >= '9999-12-31' THEN 1 ELSE 0 END) FROM sales.orders; +-- date_window +SELECT * FROM (SELECT status, COUNT(*) AS cnt FROM sales.orders WHERE paid_at >= DATE '2026-08-27' AND paid_at < DATE '9999-12-31' AND status IS NOT NULL GROUP BY status ORDER BY cnt DESC, status) WHERE ROWNUM <= 21; +-- day_volume +SELECT TRUNC(paid_at) AS d, COUNT(*) AS n FROM sales.orders WHERE paid_at >= DATE '2026-08-27' AND paid_at < DATE '9999-12-31' GROUP BY TRUNC(paid_at) ORDER BY d; +-- sample +SELECT * FROM (SELECT status, COUNT(*) AS cnt FROM sales.orders SAMPLE BLOCK (1) WHERE status IS NOT NULL GROUP BY status ORDER BY cnt DESC, status) WHERE ROWNUM <= 21; diff --git a/backend/tests/fixtures/dialect/postgresql.sql b/backend/tests/fixtures/dialect/postgresql.sql new file mode 100644 index 00000000..1416da24 --- /dev/null +++ b/backend/tests/fixtures/dialect/postgresql.sql @@ -0,0 +1,27 @@ +-- 剖析与取值观察的 SQL 金样：postgresql（tests/test_profile_dialect_golden.py 生成，改动要人审） +-- stats +SELECT c.reltuples FROM pg_catalog.pg_class c JOIN pg_catalog.pg_namespace n ON n.oid = c.relnamespace WHERE n.nspname = 'sales' AND c.relname = 'orders'; +-- stats_quoted +SELECT c.reltuples FROM pg_catalog.pg_class c JOIN pg_catalog.pg_namespace n ON n.oid = c.relnamespace WHERE n.nspname = current_schema() AND c.relname = 'visits'; +-- bounded +SELECT COUNT(*) AS n FROM (SELECT 1 AS one FROM sales.orders LIMIT 100001) s; +-- distinct +SELECT DISTINCT store_code FROM sales.orders WHERE store_code IS NOT NULL LIMIT 2000; +-- distinct_capped +SELECT DISTINCT store_code FROM (SELECT store_code FROM sales.orders WHERE store_code IS NOT NULL LIMIT 100000) s LIMIT 2000; +-- match +SELECT COUNT(DISTINCT code) AS n FROM stores WHERE code IN ('12', '7'); +-- unique +SELECT COUNT("level") AS n, COUNT(DISTINCT "level") AS d FROM "visits"; +-- value_counts +SELECT status, COUNT(*) AS cnt FROM sales.orders WHERE status IS NOT NULL GROUP BY status ORDER BY cnt DESC, status LIMIT 21; +-- value_counts_capped +SELECT status, COUNT(*) AS cnt FROM (SELECT status FROM sales.orders WHERE status IS NOT NULL LIMIT 100000) s WHERE status IS NOT NULL GROUP BY status ORDER BY cnt DESC, status LIMIT 21; +-- min_max +SELECT MIN(CASE WHEN paid_at >= DATE '0001-01-02' AND paid_at < DATE '9999-12-31' THEN paid_at END), MAX(CASE WHEN paid_at >= DATE '0001-01-02' AND paid_at < DATE '9999-12-31' THEN paid_at END), SUM(CASE WHEN paid_at < DATE '0001-01-02' OR paid_at >= DATE '9999-12-31' THEN 1 ELSE 0 END), MIN(CASE WHEN biz_day >= '0001-01-02' AND biz_day < '9999-12-31' THEN biz_day END), MAX(CASE WHEN biz_day >= '0001-01-02' AND biz_day < '9999-12-31' THEN biz_day END), SUM(CASE WHEN biz_day < '0001-01-02' OR biz_day >= '9999-12-31' THEN 1 ELSE 0 END) FROM sales.orders; +-- date_window +SELECT status, COUNT(*) AS cnt FROM sales.orders WHERE paid_at >= DATE '2026-08-27' AND paid_at < DATE '9999-12-31' AND status IS NOT NULL GROUP BY status ORDER BY cnt DESC, status LIMIT 21; +-- day_volume +SELECT CAST(paid_at AS DATE) AS d, COUNT(*) AS n FROM sales.orders WHERE paid_at >= DATE '2026-08-27' AND paid_at < DATE '9999-12-31' GROUP BY CAST(paid_at AS DATE) ORDER BY d; +-- sample +SELECT status, COUNT(*) AS cnt FROM sales.orders TABLESAMPLE SYSTEM (1) WHERE status IS NOT NULL GROUP BY status ORDER BY cnt DESC, status LIMIT 21; diff --git a/backend/tests/fixtures/dialect/sqlite.sql b/backend/tests/fixtures/dialect/sqlite.sql new file mode 100644 index 00000000..8b10405f --- /dev/null +++ b/backend/tests/fixtures/dialect/sqlite.sql @@ -0,0 +1,27 @@ +-- 剖析与取值观察的 SQL 金样：sqlite（tests/test_profile_dialect_golden.py 生成，改动要人审） +-- stats +-- （这种方言没有这类查询） +-- stats_quoted +-- （这种方言没有这类查询） +-- bounded +SELECT COUNT(*) AS n FROM (SELECT 1 AS one FROM sales.orders LIMIT 100001) s; +-- distinct +SELECT DISTINCT store_code FROM sales.orders WHERE store_code IS NOT NULL LIMIT 2000; +-- distinct_capped +SELECT DISTINCT store_code FROM (SELECT store_code FROM sales.orders WHERE store_code IS NOT NULL LIMIT 100000) s LIMIT 2000; +-- match +SELECT COUNT(DISTINCT code) AS n FROM stores WHERE code IN ('12', '7'); +-- unique +SELECT COUNT("level") AS n, COUNT(DISTINCT "level") AS d FROM "visits"; +-- value_counts +SELECT status, COUNT(*) AS cnt FROM sales.orders WHERE status IS NOT NULL GROUP BY status ORDER BY cnt DESC, status LIMIT 21; +-- value_counts_capped +SELECT status, COUNT(*) AS cnt FROM (SELECT status FROM sales.orders WHERE status IS NOT NULL LIMIT 100000) s WHERE status IS NOT NULL GROUP BY status ORDER BY cnt DESC, status LIMIT 21; +-- min_max +SELECT MIN(CASE WHEN paid_at >= '0001-01-02' AND paid_at < '9999-12-31' THEN paid_at END), MAX(CASE WHEN paid_at >= '0001-01-02' AND paid_at < '9999-12-31' THEN paid_at END), SUM(CASE WHEN paid_at < '0001-01-02' OR paid_at >= '9999-12-31' THEN 1 ELSE 0 END), MIN(CASE WHEN biz_day >= '0001-01-02' AND biz_day < '9999-12-31' THEN biz_day END), MAX(CASE WHEN biz_day >= '0001-01-02' AND biz_day < '9999-12-31' THEN biz_day END), SUM(CASE WHEN biz_day < '0001-01-02' OR biz_day >= '9999-12-31' THEN 1 ELSE 0 END) FROM sales.orders; +-- date_window +SELECT status, COUNT(*) AS cnt FROM sales.orders WHERE paid_at >= '2026-08-27' AND paid_at < '9999-12-31' AND status IS NOT NULL GROUP BY status ORDER BY cnt DESC, status LIMIT 21; +-- day_volume +SELECT substr(paid_at, 1, 10) AS d, COUNT(*) AS n FROM sales.orders WHERE paid_at >= '2026-08-27' AND paid_at < '9999-12-31' GROUP BY substr(paid_at, 1, 10) ORDER BY d; +-- sample +-- （这种方言没有这类查询） diff --git a/backend/tests/test_catalog_draft_facts.py b/backend/tests/test_catalog_draft_facts.py index 37f3705b..c65eff02 100644 --- a/backend/tests/test_catalog_draft_facts.py +++ b/backend/tests/test_catalog_draft_facts.py @@ -402,7 +402,7 @@ async def test_a_big_table_is_checked_on_its_first_rows(scenic): found = await catalog_profile.check_valid_filter(source, "visits", source.schema_cache["tables"]["visits"], "status = 1") assert found.sampled and found.rows == 100 - assert found.note.startswith("剖析核对：按前 100 行统计，") + assert found.note.startswith("剖析核对：按存储顺序取前 100 行统计，") async def test_unchecked_filters_are_kept_with_the_reason(scenic, tmp_path): diff --git a/backend/tests/test_catalog_profile.py b/backend/tests/test_catalog_profile.py index fee64c5f..2a04fe76 100644 --- a/backend/tests/test_catalog_profile.py +++ b/backend/tests/test_catalog_profile.py @@ -240,7 +240,7 @@ async def test_profile_verifies_name_relations_and_proposes_codes_and_business_d assert rel["to_table"] == parent assert (rel["source"], rel["status"]) == ("profile", "verified") assert rel["coverage"] == 1.0 and rel["cardinality"] == "many_to_one" - assert "数据剖析" in rel["note"] and "抽样" in rel["note"] + assert "数据剖析" in rel["note"] and "子表取了" in rel["note"] and "抽样" not in rel["note"] # 子表一侧数过 COUNT 和 COUNT(DISTINCT)、确有重复：基数是用数据核实的（A6，一对多关联的检查据此给 error） assert rel["cardinality_checked"] is True # A7：说明里不写日期，时间看这一项的 updated_at（界面按本地时间显示） @@ -318,7 +318,7 @@ async def test_wrong_name_relation_stays_proposed_with_coverage_in_note(client, rel = _relation((await _notes(client, sid, "parking_records"))["notes"], "lot_id") assert rel["status"] == "proposed" and rel["source"] == "profile" assert rel["coverage"] == 0.15 - assert "抽样覆盖率 15%" in rel["note"] and "可能不是这条关系" in rel["note"] and "20 个" in rel["note"] + assert "覆盖率 15%" in rel["note"] and "可能不是这条关系" in rel["note"] and "20 个" in rel["note"] finding = _table(body, "parking_records")["findings"][0] assert finding["status"] == "proposed" and finding["coverage"] == 0.15 # 两个日期类列：不提议业务日期，取值范围只在剖析结果里 diff --git a/backend/tests/test_catalog_profile_units.py b/backend/tests/test_catalog_profile_units.py index 4fa1472f..22933754 100644 --- a/backend/tests/test_catalog_profile_units.py +++ b/backend/tests/test_catalog_profile_units.py @@ -40,7 +40,7 @@ def test_settings_default_off_and_parse_leniently(): ({"table_max_s": 9}, "每张表时长上限"), ({"table_max_s": 601}, "每张表时长上限"), ({"query_timeout_s": 61}, "单条查询时限"), - ({"sample_size": 5}, "抽样键值数"), + ({"sample_size": 5}, "核对键值数"), ({"max_scan_rows": -1}, "整表统计行数上限"), ({"enabled": "true"}, "开启数据剖析"), ({"table_max_queries": True}, "每张表最多查询数"), @@ -143,8 +143,16 @@ def test_profile_sql_passes_readonly_guard_on_every_dialect(kind): d.unique_check_sql(tbl, col), d.value_counts_sql(tbl, col, 21, scan_cap=None), d.value_counts_sql(tbl, col, 21, scan_cap=100_000), - d.min_max_sql(tbl, [col]), + d.min_max_sql(tbl, [(col, False)]), + d.min_max_sql(tbl, [(col, True)]), + d.day_volume_sql(tbl, col, "2026-08-01"), + d.window_value_counts_sql(tbl, col, col, "2026-08-01", 21), ] + sampled = d.sample_value_counts_sql(tbl, col, 1, 21) + if kind in ("postgresql", "oracle"): + sqls.append(sampled) + else: + assert sampled is None # MySQL、SQLite 没有块采样 for sql in sqls: assert guard.check(sql, readonly=True, dialect=kind) == sql, sql diff --git a/backend/tests/test_catalog_semantics.py b/backend/tests/test_catalog_semantics.py index 25eb9aad..5501a4d8 100644 --- a/backend/tests/test_catalog_semantics.py +++ b/backend/tests/test_catalog_semantics.py @@ -365,7 +365,7 @@ async def test_profile_marks_codes_complete_only_when_the_whole_table_was_read(c capped = await make_source(scenic_db, options=_on(max_scan_rows=0, sample_size=10), draft=["visits"]) await _profile(client, capped, ["visits"]) codes = (await _notes(client, capped, "visits"))["notes"]["columns"]["status"]["codes"] - assert "按前 10 行统计" in codes["note"] and "complete" not in codes + assert "按存储顺序取前 10 行统计" in codes["note"] and "complete" not in codes # ========================================================================== diff --git a/backend/tests/test_profile_dialect_golden.py b/backend/tests/test_profile_dialect_golden.py new file mode 100644 index 00000000..2eb0615c --- /dev/null +++ b/backend/tests/test_profile_dialect_golden.py @@ -0,0 +1,178 @@ +"""剖析 SQL 的方言金样（下一轮 03 细案 C2）：没有真库时挡住方言写法的回退。 + +三层，各挡各的： + +1. **金样**：每种方言（mysql、mariadb、postgresql、oracle、sqlite）× 每类查询（统计信息、有上限的计数、取不同键值、 + 对键值、唯一性、取值分布、日期最小最大、时间窗口里的取值分布、按日行数、块采样），完整 SQL 钉在 + tests/fixtures/dialect/<方言>.sql，人审过一次；以后拼法有任何改动都会出 diff。要更新金样： + ``UPDATE_DIALECT_GOLDEN=1 pytest tests/test_profile_dialect_golden.py``，再逐行看 git diff。 +2. **sqlglot 逐条解析**：按方言读一遍（mariadb 按 mysql 读），挡住语法写坏了（少括号、关键字拼错）。 +3. **IN 两边类型族一致**：用 sqlglot 的 qualify + annotate_types，配上表结构里的类型，核对 IN 列表的字面量和被指向 + 列是同一类型族（PostgreSQL 上 varchar 父列配整数子列时，字面量必须带引号）。 + +挡不住的照实写：sqlglot 不知道真库的统计信息是不是过期、零日期驱动怎么返回、Oracle 小写对象是否存在、排序规则 +怎么比——这些要靠 scripts/dialect-it.sh 在本机容器里跑真库（不进 CI）。 +""" +from __future__ import annotations + +import os +from pathlib import Path + +import pytest +import sqlglot +from sqlglot import exp +from sqlglot.optimizer.annotate_types import annotate_types +from sqlglot.optimizer.qualify import qualify + +from app.data import catalog, guard +from app.data.catalog_profile import IN_CHUNK, SqlDialect, key_literals + +GOLDEN = Path(__file__).parent / "fixtures" / "dialect" +DIALECTS = ("mysql", "mariadb", "postgresql", "oracle", "sqlite") +#: sqlglot 的方言名 +READ = {"mysql": "mysql", "mariadb": "mysql", "postgresql": "postgres", "oracle": "oracle", "sqlite": "sqlite"} + +#: 夹具表结构：一张带 schema 的订单表、一张门店表（编码是文字），Oracle 的一张小写带引号的表和一个小写带引号的列 +META = { + "orders": {"schema": "sales", "columns": [ + {"name": "id", "type": "INTEGER"}, {"name": "store_code", "type": "INTEGER"}, + {"name": "status", "type": "CHAR(2)"}, {"name": "paid_at", "type": "TIMESTAMP"}, + {"name": "biz_day", "type": "VARCHAR(10)"}, {"name": "level", "type": "INTEGER"}], + "primary_key": ["id"]}, + "stores": {"schema": None, "columns": [{"name": "code", "type": "VARCHAR(10)"}], "primary_key": ["code"]}, + "visits": {"schema": None, "quote_name": True, "columns": [ + {"name": "id", "type": "INTEGER"}, {"name": "level", "type": "INTEGER", "quote_name": True}], + "primary_key": ["id"]}, +} + + +def _queries(kind: str) -> list[tuple[str, str | None]]: + """一种方言的全部剖析、观察查询：(名字, SQL)。没有这种查询（SQLite 没有统计信息、MySQL 没有块采样）的 SQL 为 None。""" + d = SqlDialect(kind) + orders, stores, visits = d.table(META["orders"], "orders"), d.table(META["stores"], "stores"), \ + d.table(META["visits"], "visits") + status, code = d.column(META["orders"], "status"), d.column(META["stores"], "code") + level = d.column(META["visits"], "level") + keys, _ = key_literals([12, 7], "number", parent="text", child_type="INTEGER") + out: list[tuple[str, str | None]] = [ + ("stats", d.stats_sql(META["orders"], "orders")), + ("stats_quoted", d.stats_sql(META["visits"], "visits")), + ("bounded", d.bounded_count_sql(orders, 100_001)), + ("distinct", d.distinct_sample_sql(orders, d.column(META["orders"], "store_code"), 2000, scan_cap=None)), + ("distinct_capped", d.distinct_sample_sql(orders, d.column(META["orders"], "store_code"), 2000, + scan_cap=100_000)), + ("match", d.match_count_sql(stores, code, keys)), + ("unique", d.unique_check_sql(visits, level)), + ("value_counts", d.value_counts_sql(orders, status, 21, scan_cap=None)), + ("value_counts_capped", d.value_counts_sql(orders, status, 21, scan_cap=100_000)), + ("min_max", d.min_max_sql(orders, [(d.column(META["orders"], "paid_at"), False), + (d.column(META["orders"], "biz_day"), True)])), + ("date_window", d.window_value_counts_sql(orders, status, d.column(META["orders"], "paid_at"), + "2026-08-27", 21)), + ("day_volume", d.day_volume_sql(orders, d.column(META["orders"], "paid_at"), "2026-08-27")), + ("sample", d.sample_value_counts_sql(orders, status, 1, 21)), + ] + return out + + +def _render(kind: str) -> str: + lines = [f"-- 剖析与取值观察的 SQL 金样：{kind}（tests/test_profile_dialect_golden.py 生成，改动要人审）"] + for name, sql in _queries(kind): + lines.append(f"-- {name}") + lines.append(f"{sql};" if sql is not None else "-- （这种方言没有这类查询）") + return "\n".join(lines) + "\n" + + +@pytest.mark.parametrize("kind", DIALECTS) +def test_profile_sql_matches_the_golden_file(kind): + path = GOLDEN / f"{kind}.sql" + text = _render(kind) + if os.environ.get("UPDATE_DIALECT_GOLDEN") == "1": + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(text, encoding="utf-8") + assert path.read_text(encoding="utf-8") == text, f"{kind} 的剖析 SQL 和金样不一致：看 diff，确认无误后重新生成金样" + + +@pytest.mark.parametrize("kind", DIALECTS) +def test_every_profile_query_parses_and_passes_the_guard(kind): + for name, sql in _queries(kind): + if sql is None: + continue + parsed = sqlglot.parse(sql, read=READ[kind], error_level=sqlglot.ErrorLevel.RAISE) + assert len(parsed) == 1 and parsed[0] is not None, (kind, name) + assert guard.check(sql, readonly=True, dialect=kind) == sql, (kind, name) + + +@pytest.mark.parametrize("kind", ["postgresql", "oracle", "mysql"]) +def test_observation_queries_use_the_right_day_and_sample_functions(kind): + """按天截断和块采样各家写法不同，解析成语法树再看用的是哪个函数，不只比字符串。""" + sql = dict(_queries(kind)) + day = sqlglot.parse_one(sql["day_volume"], read=READ[kind]) + if kind == "oracle": + assert any(isinstance(n, (exp.DateTrunc, exp.TimestampTrunc, exp.Anonymous)) or n.key == "trunc" + for n in day.walk()), sql["day_volume"] + assert "TRUNC(paid_at)" in sql["day_volume"] + elif kind == "postgresql": + assert any(isinstance(n, exp.Cast) for n in day.walk()) + else: + assert "DATE(paid_at)" in sql["day_volume"] + if kind == "mysql": + assert sql["sample"] is None + else: + sampled = sqlglot.parse_one(sql["sample"], read=READ[kind]) + assert any(isinstance(n, exp.TableSample) for n in sampled.walk()), sql["sample"] + + +# ---------------------------------------------------------------- IN 两边的类型族 + + +_FAMILY_OF = [(exp.DataType.TEXT_TYPES, "text"), (exp.DataType.NUMERIC_TYPES, "number"), + (exp.DataType.TEMPORAL_TYPES, "time")] + + +def _sg_family(t: exp.DataType | None) -> str | None: + if t is None: + return None + for group, name in _FAMILY_OF: + if t.this in group: + return name + return None + + +def _in_families(sql: str, kind: str, schema: dict[str, dict[str, str]]) -> list[tuple[str | None, set[str | None]]]: + """sql 里每个 IN：(被比较的列的类型族, 字面量的类型族集合)，按 annotate_types 配上表结构的类型推出来。""" + tree = qualify(sqlglot.parse_one(sql, read=READ[kind]), schema=schema, dialect=READ[kind], + validate_qualify_columns=False) + tree = annotate_types(tree, schema=schema, dialect=READ[kind]) + return [(_sg_family(node.this.type), {_sg_family(x.type) for x in node.expressions}) + for node in tree.find_all(exp.In)] + + +@pytest.mark.parametrize("kind", DIALECTS) +@pytest.mark.parametrize("parent_type, child_values, child_type", [ + ("VARCHAR(10)", [12, 7, 30], "INTEGER"), # 父列文字、子列数：字面量要带引号 + ("INTEGER", ["12", "0007", "A1"], "VARCHAR(10)"), # 父列数、子列文字：只收整数写法，写成数字 + ("VARCHAR(10)", ["A ", "BC"], "CHAR(2)"), # 定长子列：去掉补齐的空格 + ("INTEGER", [1, 2], "INTEGER"), +]) +def test_in_list_literals_have_the_parent_columns_family(kind, parent_type, child_values, child_type): + d = SqlDialect(kind) + parent = {"name": "code", "type": parent_type} + literals, _ = key_literals(child_values, None, parent=catalog.column_family(parent), child_type=child_type) + assert literals + sql = d.match_count_sql("stores", d.quote("code"), literals[:IN_CHUNK]) + found = _in_families(sql, kind, {"stores": {"code": parent_type}}) + assert found, sql + for column_family, literal_families in found: + assert literal_families == {column_family}, (kind, sql, column_family, literal_families) + + +def test_the_old_way_would_mismatch_on_postgres(): + """对照：以前按子表取回的值写字面量，varchar 父列配整数子列在 sqlglot 推出的类型上就两边不同——PostgreSQL 真库上 + 这是「operator does not exist: character varying = integer」。""" + from app.data.catalog_profile import sql_literal + + old = [lit for v in (12, 7) if (lit := sql_literal(v, "number"))] + sql = SqlDialect("postgresql").match_count_sql("stores", "code", old) + [(column_family, literal_families)] = _in_families(sql, "postgresql", {"stores": {"code": "VARCHAR(10)"}}) + assert column_family == "text" and literal_families == {"number"} diff --git a/backend/tests/test_profile_dialect_units.py b/backend/tests/test_profile_dialect_units.py new file mode 100644 index 00000000..db01b746 --- /dev/null +++ b/backend/tests/test_profile_dialect_units.py @@ -0,0 +1,296 @@ +"""剖析与探查的方言修正（下一轮 03 细案 C1）：不连库的部分。 + +- 列类型按方言编译的完整写法：类型族、日期列、码值候选都认得 ENUM / SET、VARCHAR2(20 CHAR)、INTERVAL 这些写法； +- IN 列表的字面量按被指向列的类型族生成（PG 上文字列配整数子列会直接报错，Oracle 会把父列转成数报 ORA-01722）； +- Oracle 小写带引号的对象（quote_name）强制加引号、不转大写； +- 统计信息估出的行数不超过整表统计上限时再数一次确认； +- 日期列的最小值、最大值排除零日期和哨兵日期，照实写排除了几个； +- 取值观察要用的三个方言方法（date_literal / day_trunc / sample_from）； +- 说明文字不再说「抽样」：大表看的是按存储顺序的前 N 行。 +""" +from __future__ import annotations + +import asyncio +import datetime as _dt +from typing import Any + +import pytest +from sqlalchemy import VARCHAR +from sqlalchemy.dialects import mysql, oracle, postgresql, sqlite +from sqlalchemy.sql.elements import quoted_name + +from app.data import catalog, catalog_profile, introspect +from app.data.catalog_profile import ProfileSettings, SqlDialect, TableSize +from app.data.engine import QueryResult + +DIALECTS = ("mysql", "mariadb", "postgresql", "oracle", "sqlite") + + +# ---------------------------------------------------------------- 类型的写法 + + +@pytest.mark.parametrize("type_name, family", [ + ("ENUM('paid','refund')", "text"), + ("ENUM('int','date')", "text"), # 取值里的词不能让它变成数字或时间 + ("SET('a','b')", "text"), + ("VARCHAR2(20 CHAR)", "text"), + ("NVARCHAR2(10)", "text"), + ("VARCHAR(20) COLLATE utf8mb4_bin", "text"), + ("NUMBER(10, 0)", "number"), + ("INTEGER(11)", "number"), + ("BIGINT UNSIGNED", "number"), + ("DOUBLE PRECISION", "number"), + ("TIMESTAMP WITH TIME ZONE", "time"), + ("TIMESTAMP(3)", "time"), + ("INTERVAL DAY(2) TO SECOND(6)", None), # 时长不是数也不是时刻；以前按「INT」认成了数字 + ("order_status", None), + ("", None), +]) +def test_type_family_reads_dialect_full_types(type_name, family): + assert catalog._type_family(type_name) == family + + +def test_type_family_of_a_column_uses_its_enum_values(): + """PostgreSQL 的枚举类型编译出来是类型名（order_status），只有取值列表能说明它是文字。""" + assert catalog.column_family({"name": "status", "type": "order_status", "enum_values": ["a", "b"]}) == "text" + assert catalog.column_family({"name": "n", "type": "INTEGER"}) == "number" + assert catalog.column_family({"name": "x"}) is None + + +def test_enum_values_never_make_a_date_column(): + assert not catalog.is_date_column("status", "ENUM('date','time')") + assert catalog.is_date_column("paid_at", "TIMESTAMP WITH TIME ZONE") + assert catalog.is_date_column("biz_dt", "VARCHAR2(10 CHAR)") # 文字存日期、名字像日期 + assert not catalog.is_date_column("gap", "INTERVAL DAY(2) TO SECOND(6)") + + +@pytest.mark.parametrize("col, ok", [ + ({"type": "VARCHAR2(20 CHAR)"}, True), + ({"type": "VARCHAR2(200 CHAR)"}, False), + ({"type": "CHAR(2)"}, True), + ({"type": "ENUM('paid','refund')"}, True), + ({"type": "SET('a','b')"}, True), + ({"type": "order_status", "enum_values": ["a", "b"]}, True), + ({"type": "NUMBER(10, 0)"}, True), + ({"type": "NUMBER(10, 2)"}, False), + ({"type": "TIMESTAMP(3)"}, False), + ({"type": "INTERVAL DAY(2) TO SECOND(6)"}, False), +]) +def test_code_type_accepts_enum_and_char_semantics(col, ok): + assert catalog_profile._code_type_ok(col.get("type"), col.get("enum_values")) is ok + + +def test_enum_columns_are_code_candidates_whatever_their_name(): + """ENUM 的取值列表有限，直接算码值候选：pay_way 这种名字不像码值的列以前被漏掉。""" + meta = {"columns": [{"name": "id", "type": "INTEGER"}, {"name": "pay_way", "type": "ENUM('cash','card')", + "enum_values": ["cash", "card"]}, + {"name": "pay_note", "type": "VARCHAR(20)"}], + "primary_key": ["id"]} + assert catalog_profile._code_columns(meta, {}) == ["pay_way"] + + +# ---------------------------------------------------------------- 探查：类型写法、取值列表、引号、索引首列 + + +@pytest.mark.parametrize("dialect, col_type, text", [ + (mysql.dialect(), mysql.ENUM("paid", "refund"), "ENUM('paid','refund')"), + (mysql.dialect(), mysql.TINYINT(display_width=1), "TINYINT(1)"), + (oracle.dialect(), oracle.VARCHAR2(20), "VARCHAR2(20 CHAR)"), + # Oracle 的 VARCHAR2 反射回来是通用的 VARCHAR（容器集成测试实测）：照数据字典写成 VARCHAR2 + (oracle.dialect(), VARCHAR(10), "VARCHAR2(10 CHAR)"), + (oracle.dialect(), oracle.NUMBER(10, 0), "NUMBER(10, 0)"), + (postgresql.dialect(), postgresql.ENUM("a", "b", name="order_status"), "order_status"), + (sqlite.dialect(), sqlite.DATETIME(), "DATETIME"), +]) +def test_reflected_type_is_compiled_for_the_dialect(dialect, col_type, text): + assert introspect.type_text(col_type, dialect) == text + + +def test_type_that_cannot_compile_falls_back_to_str(): + from sqlalchemy.types import NullType + + assert introspect.type_text(NullType(), sqlite.dialect()) == "NULL" + assert introspect.type_text(None, sqlite.dialect()) == "" + + +def test_column_entry_records_enum_values_and_oracle_quoted_names(): + ora = oracle.dialect() + entry = introspect.column_entry({"name": quoted_name("level", quote=True), "type": oracle.VARCHAR2(10), + "nullable": True, "comment": None}, ora) + assert entry == {"name": "level", "type": "VARCHAR2(10 CHAR)", "nullable": True, "comment": None, + "quote_name": True} + plain = introspect.column_entry({"name": "status", "type": mysql.ENUM("a", "b"), "nullable": False}, mysql.dialect()) + assert plain["enum_values"] == ["a", "b"] and "quote_name" not in plain + members = introspect.column_entry({"name": "tags", "type": mysql.SET("x", "y")}, mysql.dialect()) + assert members["enum_values"] == ["x", "y"] and members["type"] == "SET('x','y')" + + +def test_index_first_columns_are_recorded(): + indexes = [{"column_names": ["paid_at", "id"]}, {"column_names": [None]}, {"column_names": ["store_id"]}, + {"column_names": ["paid_at"]}] + assert introspect.index_first_columns(indexes, ["id"]) == ["id", "paid_at", "store_id"] + + +# ---------------------------------------------------------------- IN 列表的字面量按被指向列的类型族 + + +def test_key_literals_follow_the_parent_family(): + lits, mismatched = catalog_profile.key_literals([12, 7], "number", parent="text", child_type="INTEGER") + assert lits == ["'12'", "'7'"] and mismatched == 0 + lits, mismatched = catalog_profile.key_literals(["12", "A1", "-3"], "text", parent="number", + child_type="VARCHAR(10)") + assert lits == ["12", "-3"] and mismatched == 1 + lits, mismatched = catalog_profile.key_literals([1, 2], "number", parent="time", child_type="INTEGER") + assert lits == [] and mismatched == 2 + # 父列类型不知道：照子表取回的值的类型写（以前的做法） + assert catalog_profile.key_literals([3, "x"], None, parent=None, child_type=None) == (["3", "'x'"], 0) + + +def test_char_keys_lose_their_padding(): + """PG、Oracle 的 CHAR(n) 取回来带尾随空格：拼成 'A ' 去和 VARCHAR 父列比就对不上。""" + lits, _ = catalog_profile.key_literals(["A ", "BC"], "text", parent="text", child_type="CHAR(2)") + assert lits == ["'A'", "'BC'"] + lits, _ = catalog_profile.key_literals(["A "], "text", parent="text", child_type="VARCHAR(2)") + assert lits == ["'A '"] + + +# ---------------------------------------------------------------- Oracle 小写带引号的对象 + + +def test_oracle_quote_name_forces_quotes_and_keeps_case(): + ora = SqlDialect("oracle") + meta = {"schema": "sales", "quote_name": True, + "columns": [{"name": "level", "type": "VARCHAR2(10 CHAR)", "quote_name": True}, + {"name": "amount", "type": "NUMBER(10, 2)"}]} + assert ora.table(meta, "visits") == 'sales."visits"' + assert ora.column(meta, "level") == '"level"' and ora.column(meta, "amount") == "amount" + stats = ora.stats_sql(meta, "visits") + assert "TABLE_NAME = 'visits'" in stats and "OWNER = 'SALES'" in stats + # 没有 quote_name 的照旧：不加引号，Oracle 自己转大写 + assert ora.table({"schema": None}, "visits") == "visits" + assert "TABLE_NAME = 'VISITS'" in ora.stats_sql({}, "visits") + + +# ---------------------------------------------------------------- 统计信息过期：再数一次确认 + + +def _result(value: Any) -> QueryResult: + return QueryResult(columns=["n"], rows=[[value]], row_count=1, truncated=False, elapsed_ms=1, sql="") + + +def _runner(answers: list[Any], seen: list[str]): + async def run(sql: str, **_kw: Any) -> QueryResult: + seen.append(sql) + return _result(answers.pop(0)) + return run + + +@pytest.mark.parametrize("stats, counted, expected", [ + (500, 500, TableSize(500, "count")), # 统计信息说小、数过也小：准数 + (500, 100_001, TableSize(None, "count", at_least=100_001)), # 统计信息过期：其实是大表 + (5_000_000, None, TableSize(5_000_000, "stats")), # 统计信息说大：不必再数 +]) +def test_small_stats_are_confirmed_with_a_bounded_count(stats, counted, expected): + seen: list[str] = [] + answers = [stats] + ([counted] if counted is not None else []) + size, skips = asyncio.run(catalog_profile.estimate_size(_runner(answers, seen), SqlDialect("postgresql"), + {"schema": None}, "visits", max_scan_rows=100_000)) + assert size == expected and skips == [] + assert len(seen) == (2 if counted is not None else 1) + if counted is not None: + assert "LIMIT 100001" in seen[1] + + +# ---------------------------------------------------------------- 日期列：排除零日期和哨兵日期 + + +@pytest.mark.parametrize("kind", DIALECTS) +def test_min_max_excludes_invalid_dates_and_counts_them(kind): + d = SqlDialect(kind) + sql = d.min_max_sql("t", [("paid_at", False), ("biz_day", True)]) + # 日期类型的列按日期字面量比，文字存的日期按字符串比（PG 上 varchar 和 date 比较直接报错） + lo, hi = d.date_literal("0001-01-02"), d.date_literal("9999-12-31") + assert f"paid_at >= {lo}" in sql and f"paid_at < {hi}" in sql + assert "biz_day >= '0001-01-02'" in sql and "biz_day < '9999-12-31'" in sql + assert sql.count("SUM(CASE WHEN") == 2 + + +def _dates_cx(row: list[Any], kind: str = "mysql"): + settings = ProfileSettings(enabled=True) + source = type("S", (), {"id": "s1", "name": "shop", "kind": kind, "options": {}})() + job = catalog_profile.ProfileContext(source=source, settings=settings, dialect=SqlDialect(kind), + runner=None, tables={"orders": {"columns": [ + {"name": "paid_at", "type": "DATETIME"}]}}, masked=set()) + job.sizes["orders"] = TableSize(10, "count") + result = catalog_profile.TableResult(table="orders") + cx = catalog_profile._TableCx(job, "orders", result, catalog_profile.TableHooks()) + + async def run(sql: str, **_kw: Any) -> QueryResult: + return QueryResult(columns=["a", "b", "c"], rows=[row], row_count=1, truncated=False, elapsed_ms=1, sql=sql) + cx.run = run # type: ignore[method-assign] + return cx, result + + +def test_dates_note_says_how_many_invalid_values_were_left_out(): + cx, result = _dates_cx(["2026-01-03 08:00:00", "2026-09-30 21:00:00", 4]) + out = catalog_profile._TableOutcome() + asyncio.run(catalog_profile._check_dates(cx, "orders", ["paid_at"], 1, out, result)) + assert result.date_ranges == [{"column": "paid_at", "min": "2026-01-03 08:00:00", "max": "2026-09-30 21:00:00", + "invalid": 4}] + assert "另有 4 个无效日期" in out.business_date.note and "0000-00-00" in out.business_date.note + + +def test_only_invalid_dates_is_no_data(): + cx, result = _dates_cx([None, None, 9]) + with pytest.raises(catalog_profile._Skip) as e: + asyncio.run(catalog_profile._check_dates(cx, "orders", ["paid_at"], 1, catalog_profile._TableOutcome(), + result)) + assert e.value.reason == "no_data" and "9 个无效日期" in e.value.detail + + +# ---------------------------------------------------------------- 取值观察要用的方言方法 + + +@pytest.mark.parametrize("kind, literal, day, sample", [ + ("mysql", "DATE '2026-09-01'", "DATE(paid_at)", None), + ("mariadb", "DATE '2026-09-01'", "DATE(paid_at)", None), + ("postgresql", "DATE '2026-09-01'", "CAST(paid_at AS DATE)", "orders TABLESAMPLE SYSTEM (1.5)"), + ("oracle", "DATE '2026-09-01'", "TRUNC(paid_at)", "orders SAMPLE BLOCK (1.5)"), + ("sqlite", "'2026-09-01'", "substr(paid_at, 1, 10)", None), +]) +def test_observation_dialect_methods(kind, literal, day, sample): + d = SqlDialect(kind) + assert d.date_literal(_dt.date(2026, 9, 1)) == literal == d.date_literal("2026-09-01") + assert d.date_literal("2026-09-01", text=True) == "'2026-09-01'" + assert d.day_trunc("paid_at") == day + assert d.sample_from("orders", 1.5) == sample + + +@pytest.mark.parametrize("bad", ["2026-9-1", "2026-09-01'; --", "0000-00-00", 20260901, None]) +def test_date_literal_refuses_anything_but_a_real_date(bad): + with pytest.raises(ValueError): + SqlDialect("postgresql").date_literal(bad) + + +@pytest.mark.parametrize("bad", [0, -1, 100, 101, float("nan")]) +def test_sample_percent_must_be_between_0_and_100(bad): + with pytest.raises(ValueError): + SqlDialect("postgresql").sample_from("orders", bad) + + +# ---------------------------------------------------------------- 说明文字 + + +def test_notes_say_storage_order_not_sampling(): + note = catalog_profile._relation_note(sample=2000, matched=2000, coverage=1.0, scan_cap=100_000, skipped_values=0, + mismatched=0, parent=(True, "主键"), child=(False, "count"), holds=True, + collation=None) + assert "按存储顺序只看前 100000 行" in note and "抽样" not in note + assert catalog_profile.FILTER_SAMPLED.format(rows="100", share="1%").startswith("剖析核对：按存储顺序取前 100 行统计") + + +def test_mysql_notes_mention_the_collation(): + note = catalog_profile._relation_note(sample=10, matched=10, coverage=1.0, scan_cap=None, skipped_values=0, + mismatched=0, parent=(True, "主键"), child=(False, "count"), holds=True, + collation=catalog_profile.MYSQL_COLLATION_NOTE) + assert "排序规则" in note and "不分大小写" in note diff --git a/frontend/src/lib/terms.ts b/frontend/src/lib/terms.ts index 51fb6dba..ed0661cf 100644 --- a/frontend/src/lib/terms.ts +++ b/frontend/src/lib/terms.ts @@ -2771,7 +2771,7 @@ export const SCHEMA_PARTIAL_TEXT = { // - 剖析是后台任务：逐张表进行，每张表剖析完就写进目录；查询一条一条地发，可以随时停止。关掉对话框照常进行。 // 目录只在整张表结束时写一次：停止、关停服务时正在做的那张来得及收尾，做完的项照常写进去；进程被杀时那张表没来得及收尾， // 这次做完的项一项都没写进去。说停下以后写进了什么，按逐表的状态照实说（cancelledNote / interruptedNote），不一律说「已写进」。 -// - 预算按表算：「每张表最多查询数」「每张表时长上限」用完只停这张表，下一张表重新计数；另有单条查询时限、抽样键值数、 +// - 预算按表算：「每张表最多查询数」「每张表时长上限」用完只停这张表，下一张表重新计数；另有单条查询时限、核对键值数、 // 整表统计行数上限。标签和服务端校验报错里的叫法一致（catalog_profile._NUMBER_FIELDS），保存被拒时按机读码认出是哪一项。 // - 没有整次总量上限：任务自己在连续失败、多数超时时停下（止损），写明原因。 // - 「没查成」≠ 结论错了：因为次数、时长、超时、出错没查成的项保留剖析以前的结论，可以「重试没做完的表」。 @@ -2783,7 +2783,7 @@ export const PROFILE_FIELD_LABEL: Record<'enabled' | CatalogProfileNumberKey, st table_max_queries: '每张表最多查询数', table_max_s: '每张表时长上限', query_timeout_s: '单条查询时限', - sample_size: '抽样键值数', + sample_size: '核对键值数', max_scan_rows: '整表统计行数上限', } @@ -2799,7 +2799,7 @@ export const PROFILE_FIELD_HINT: Record = { table_max_queries: '剖析一张表最多发出的查询条数，用完只停这张表，下一张表重新计数', table_max_s: '剖析一张表最多用的时长，到时只停这张表，下一张表重新计时', query_timeout_s: '超过即中断这条查询；不超过数据源自身的查询时限，也不超过这张表剩下的时长', - sample_size: '核对一条关联关系时，从本表抽取的不同键值个数', + sample_size: '核对一条关联关系时，从本表取的不同键值个数：取的是数据库先给出的那些，不是随机抽样', max_scan_rows: '不超过此行数的表才做去重计数、最小值和最大值等整表统计', } @@ -2863,7 +2863,7 @@ export const PROFILE_TEXT = { budgetPerTable: (queries: number, secs: number) => `每张表最多 ${formatNumber(queries)} 条查询、${formatNumber(secs)} 秒，用完只停这张表，下一张表重新计数`, budgetQuery: (secs: number) => `单条查询不超过 ${formatNumber(secs)} 秒`, - budgetSample: (n: number) => `每条关联关系抽样 ${formatNumber(n)} 个键值`, + budgetSample: (n: number) => `每条关联关系取 ${formatNumber(n)} 个不同键值核对`, budgetScan: (n: number) => (n > 0 ? `行数超过 ${formatNumber(n)} 的表不做整表统计` : '不做整表统计'), /** 上限的估算：表数 × 每张表的上限 */ budgetTotal: (tables: number, queries: number, span: string) => @@ -3002,7 +3002,7 @@ export const PROFILE_TEXT = { relationLowCoverage: '覆盖率不足，可能不是这条关系', relationNotUnique: '被指向列未核实唯一', coverage: (pct: string) => `覆盖率 ${pct}`, - sampled: (sample: number, matched: number) => `抽样 ${formatNumber(sample)} 个键值，对上 ${formatNumber(matched)} 个`, + sampled: (sample: number, matched: number) => `取了 ${formatNumber(sample)} 个不同键值，对上 ${formatNumber(matched)} 个`, cardinalityUnknown: '基数未核实', codeRows: (n: number) => `${formatNumber(n)} 行`, codesSummary: (values: number, rows: number) => `${formatNumber(values)} 个取值，共 ${formatNumber(rows)} 行`, @@ -3012,10 +3012,12 @@ export const PROFILE_TEXT = { dateProposal: (col: string) => `提议按 ${col} 作为业务日期`, dateSpan: (min: string, max: string) => `取值从 ${min} 到 ${max}`, dateRanges: '日期列的取值范围', + /** 日期列的最小值、最大值不计入的值（catalog_profile.VALID_DATE_FROM）：零日期、0001-01-01、9999-12-31、无穷大 */ + dateInvalid: (n: number) => `另有 ${formatNumber(n)} 个无效日期未计入（零日期或占位日期）`, skippedTitle: (n: number) => `跳过 ${formatNumber(n)} 项`, moreSkipped: (n: number) => `…另有 ${formatNumber(n)} 项`, queriesTitle: (n: number) => `看发过的查询（${formatNumber(n)} 条）`, - queriesHint: 'SQL 里的取值都换成了「?」：抽样到的键值不保存在 AgentLab 中；遮罩的列不出现在任何查询里', + queriesHint: 'SQL 里的取值都换成了「?」：取到的键值不保存在 AgentLab 中；遮罩的列不出现在任何查询里', queryKind: { row_estimate: '行数估算', relation: '关系', codes: '码值', date: '日期' } as Record, queryOutcome: { ok: '成功', timeout: '超时', error: '失败', rejected: '未通过只读守卫' } as Record, queryRows: (n: number) => `${formatNumber(n)} 行`, diff --git a/frontend/src/pages/catalog/ProfileDialog.tsx b/frontend/src/pages/catalog/ProfileDialog.tsx index 321ad9c3..b14641fc 100644 --- a/frontend/src/pages/catalog/ProfileDialog.tsx +++ b/frontend/src/pages/catalog/ProfileDialog.tsx @@ -780,6 +780,10 @@ export function ProfileTaskBody({ detail: d, mode, filled = {}, focusAfter = nul {PT.dateProposal(date.column)} ，{PT.dateSpan(date.min.slice(0, 19), date.max.slice(0, 19))} + {(() => { + const invalid = d.date_ranges.find((r) => r.column === date.column)?.invalid + return invalid ? ，{PT.dateInvalid(invalid)} : null + })()} )} @@ -788,6 +792,7 @@ export function ProfileTaskBody({ detail: d, mode, filled = {}, focusAfter = nul {d.date_ranges.map((r) => (
  • {r.column}：{PT.dateSpan(r.min.slice(0, 19), r.max.slice(0, 19))} + {r.invalid ? `，${PT.dateInvalid(r.invalid)}` : null}
    • ))} diff --git a/frontend/src/pages/catalog/ProfileSettings.tsx b/frontend/src/pages/catalog/ProfileSettings.tsx index ddd9a3ec..91f13ee1 100644 --- a/frontend/src/pages/catalog/ProfileSettings.tsx +++ b/frontend/src/pages/catalog/ProfileSettings.tsx @@ -14,7 +14,7 @@ import type { ProfileForm } from './profile' // 数据剖析的设置：开关、五项预算、风险说明。数据源编辑框里是其中一段；导入表格的源没有编辑框，数据目录页在 // 剖析被拒（未开启）时也要就地打开它，所以另有一个只改这一项的弹窗（options 的其余键原样带上）。 // 剖析会对业务库发查询，开关默认关着；开启后五项预算才展开，风险说明写的是此刻填的数。 -// 预算按表算：每张表最多查询数、每张表时长上限（用完只停这张表），另有单条查询时限、抽样键值数、整表统计行数上限； +// 预算按表算：每张表最多查询数、每张表时长上限（用完只停这张表），另有单条查询时限、核对键值数、整表统计行数上限； // 没有整次剖析的总量上限（任务自己在连续失败、多数超时时停下）。 // =========================================================================== diff --git a/frontend/src/types.ts b/frontend/src/types.ts index 0bd204f6..f44637d7 100644 --- a/frontend/src/types.ts +++ b/frontend/src/types.ts @@ -3528,7 +3528,8 @@ export interface CatalogProfileTaskDetail extends CatalogProfileTaskRow { findings: CatalogProfileFinding[] skipped: CatalogProfileSkip[] /** 日期类列的取值范围 */ - date_ranges: { column: string; min: string; max: string }[] + /** invalid：零日期、占位日期这类不计入最小值和最大值的个数（没有时不给） */ + date_ranges: { column: string; min: string; max: string; invalid?: number }[] query_log: CatalogProfileQuery[] } diff --git a/scripts/check-catalog-profile.mjs b/scripts/check-catalog-profile.mjs index 66efe68e..5c7f6cde 100644 --- a/scripts/check-catalog-profile.mjs +++ b/scripts/check-catalog-profile.mjs @@ -695,7 +695,7 @@ await section('剖析 · 设置：每张表的两项预算、被拒时报错落 await sec.locator('[data-profile-enabled]').check() const labels = await sec.locator('label.label').allInnerTexts() check('开启后五项预算展开：每张表最多查询数、每张表时长上限在前，没有整次的查询次数上限、总时长上限', - labels.join('、') === '每张表最多查询数、每张表时长上限、单条查询时限、抽样键值数、整表统计行数上限', labels.join('、')) + labels.join('、') === '每张表最多查询数、每张表时长上限、单条查询时限、核对键值数、整表统计行数上限', labels.join('、')) const units = await sec.locator('[data-profile-field] span').allInnerTexts() check('……每项带单位', units.join(',') === '条,秒,秒,个,行', units.join(',')) const hints = await sec.locator('[id$="-hint"]').allInnerTexts() @@ -783,8 +783,8 @@ await section('剖析 · 开始前：范围、个数与预算', async () => { check('……写明需要剖析的口径（含上次是空表的不算）、全部不含视图', (await inBox(page, '[data-profile-scope="needed"]')).includes('从没剖析过、上次没做完，或剖析之后又出现了待核实关联关系、要取码值或日期范围的列的表。上次是空表的不算') && (await inBox(page, '[data-profile-scope="all"]')).includes('不含视图')) const budget = await inBox(page, '[data-profile-budget]') - check('预算按表写：每张表最多 40 条、60 秒，单条 10 秒，抽样、整表统计门槛，按哪个源的设置', - ['每张表最多 40 条查询、60 秒，用完只停这张表', '单条查询不超过 10 秒', '抽样 2,000 个键值', '行数超过 100,000 的表不做整表统计', `按「${NAME1}」的数据剖析设置`] + check('预算按表写：每张表最多 40 条、60 秒，单条 10 秒，核对键值数、整表统计门槛，按哪个源的设置', + ['每张表最多 40 条查询、60 秒，用完只停这张表', '单条查询不超过 10 秒', '取 2,000 个不同键值核对', '行数超过 100,000 的表不做整表统计', `按「${NAME1}」的数据剖析设置`] .every((x) => budget.includes(x)), budget) check('……按这次的表数估上限：「这次 5 张表，最多约 200 条查询、5 分钟」', (await inBox(page, '[data-profile-budget-total]')) === '这次 5 张表，最多约 200 条查询、5 分钟', await inBox(page, '[data-profile-budget-total]')) diff --git a/scripts/dialect-it.sh b/scripts/dialect-it.sh new file mode 100755 index 00000000..0b7319ec --- /dev/null +++ b/scripts/dialect-it.sh @@ -0,0 +1,62 @@ +#!/usr/bin/env bash +# 方言的容器集成测试（下一轮 03 细案 C2）：本机 docker 起 PostgreSQL 16、MySQL 8.4、MariaDB 11、Oracle Free 23 四个容器， +# 建虚构的夹具（保留字列、Oracle 小写带引号的表、CHAR 码值、MySQL ENUM、零日期 / 无穷大 / 占位日期、做旧的统计信息、 +# 超过整表统计上限的大表、带索引的日期列），用 AgentLab 自己的探查和剖析代码跑全部剖析查询和取值观察的新方法， +# 断言零报错、结论和手算的一致。 +# +# 为什么不进 CI：要拉几 GB 的镜像、Oracle 起一次要一两分钟。要用户同意才在本机跑。跑完删容器、留镜像。 +# +# 用法：scripts/dialect-it.sh 四种库都跑 +# DIALECT_IT_ONLY=pg,mysql scripts/dialect-it.sh 只跑其中几种（pg、mysql、mariadb、oracle） +# PYTHON=/path/to/python scripts/dialect-it.sh 指定装着后端依赖的 Python（缺省 python3） +# +# 夹具全是虚构的通用名字（stores、orders、events、visits），不连任何业务库。 +set -euo pipefail + +ROOT="$(cd "$(dirname "$0")/.." && pwd)" +PY="${PYTHON:-python3}" +ONLY="${DIALECT_IT_ONLY:-pg,mysql,mariadb,oracle}" +PREFIX="agentlab-dialect-it" +PASS="itpass" + +started=() +cleanup() { + for name in "${started[@]}"; do + docker rm -f "$name" >/dev/null 2>&1 || true + done +} +trap cleanup EXIT + +want() { [[ ",$ONLY," == *",$1,"* ]]; } + +run() { + local name="$PREFIX-$1"; shift + docker rm -f "$name" >/dev/null 2>&1 || true + docker run -d --name "$name" "$@" >/dev/null + started+=("$name") +} + +host_port() { docker port "$PREFIX-$1" "$2" | head -1 | sed 's/.*://'; } + +if want pg; then + run pg -e POSTGRES_PASSWORD="$PASS" -e POSTGRES_DB=it -p 127.0.0.1::5432 postgres:16 +fi +if want mysql; then + run mysql -e MYSQL_ROOT_PASSWORD="$PASS" -e MYSQL_DATABASE=it -p 127.0.0.1::3306 mysql:8.4 +fi +if want mariadb; then + run mariadb -e MARIADB_ROOT_PASSWORD="$PASS" -e MARIADB_DATABASE=it -p 127.0.0.1::3306 mariadb:11 +fi +if want oracle; then + run oracle -e ORACLE_PASSWORD="$PASS" -e APP_USER=it -e APP_USER_PASSWORD="$PASS" -p 127.0.0.1::1521 \ + gvenzl/oracle-free:23-slim +fi + +export DIALECT_IT_PASSWORD="$PASS" DIALECT_IT_ONLY="$ONLY" +want pg && export DIALECT_IT_PG_PORT="$(host_port pg 5432/tcp)" +want mysql && export DIALECT_IT_MYSQL_PORT="$(host_port mysql 3306/tcp)" +want mariadb && export DIALECT_IT_MARIADB_PORT="$(host_port mariadb 3306/tcp)" +want oracle && export DIALECT_IT_ORACLE_PORT="$(host_port oracle 1521/tcp)" + +# 等库起来由 Python 那边按连接重试（Oracle 最慢，最多等 6 分钟） +PYTHONPATH="$ROOT/backend" "$PY" "$ROOT/scripts/dialect_it.py" diff --git a/scripts/dialect_it.py b/scripts/dialect_it.py new file mode 100644 index 00000000..ff153dd1 --- /dev/null +++ b/scripts/dialect_it.py @@ -0,0 +1,388 @@ +"""方言容器集成测试的执行端（由 scripts/dialect-it.sh 起好容器后调用，不进 CI）。 + +对每种库： +1. 用管理账号建虚构夹具（见 _ddl / _rows）； +2. 用 AgentLab 自己的探查（app.data.introspect）读结构，核对类型写法、ENUM 取值、Oracle 小写带引号的名字、索引首列； +3. 用 AgentLab 自己的剖析（catalog_profile.profile_table，目录读写换成内存里的）逐表剖析，经只读连接、守卫、单条时限， + 和线上走的是同一条路； +4. 直接跑取值观察要用的新查询（按日行数、时间窗口里的取值分布、块采样）； +5. 断言：所有查询零报错（留痕里没有超时、失败、守卫拒），每个结论和下面按夹具定义手算的一致。 + +夹具的数（手算的依据）：orders 40 行，id 1–40。 +- store_code = id % 5 + 1（整数子列，指向文字主键 stores.code，'1'–'5' 都在）； +- status CHAR(2)：id 是 3 的倍数为 'A'，否则 'B'（13 / 27）；type：奇数 'web'、偶数 'shop'（20 / 20）； + level = id % 3 + 1（1: 13、2: 14、3: 13）；pay_way（只 MySQL / MariaDB，ENUM('cash','card','coupon')）：奇数 cash、偶数 card； +- paid_at = 2026-09-01 10:00 加 (id % 10) 天，id 39、40 是无效日期（MySQL 零日期、PG ±infinity、Oracle 0001-01-01 和 + 9999-12-31 23:59:59）；date（日期列）同 paid_at 的日子，id 39、40 为空； +- events 先插 10 行、收集统计信息，再插 1500 行（统计信息做旧：说 10 行，其实 1510 行，超过整表统计上限 1000）； +- Oracle 另有小写带引号的表 "visits"（"id"、"level"），3 行，level 1、1、2。 +""" +from __future__ import annotations + +import asyncio +import datetime as _dt +import os +import sys +import tempfile +import time +from collections import Counter +from types import SimpleNamespace +from typing import Any + +# AgentLab 自己的库放到临时目录：不碰开发数据目录（必须在导入 app 之前设） +os.environ["AGENTLAB_DATA_DIR"] = tempfile.mkdtemp(prefix="agentlab-dialect-it-") + +from sqlalchemy import text # noqa: E402 +from sqlalchemy.ext.asyncio import create_async_engine # noqa: E402 + +from app.core.crypto import encrypt # noqa: E402 +from app.data import catalog, catalog_profile, introspect # noqa: E402 +from app.data import engine as data_engine # noqa: E402 +from app.data.catalog_profile import ProfileSettings, SqlDialect # noqa: E402 + +PASSWORD = os.environ.get("DIALECT_IT_PASSWORD", "itpass") +SETTINGS = ProfileSettings(enabled=True, table_max_queries=40, table_max_s=60, query_timeout_s=10, sample_size=100, + max_scan_rows=1000) +N_ORDERS, N_EVENTS_OLD, N_EVENTS_NEW = 40, 10, 1500 +BASE = _dt.datetime(2026, 9, 1, 10, 0, 0) +WINDOW_FROM = "2026-09-05" + +failures: list[str] = [] + + +def check(ok: bool, what: str) -> None: + print((" ✓ " if ok else " ✗ ") + what) + if not ok: + failures.append(what) + + +# ---------------------------------------------------------------- 手算 + + +def _order(i: int) -> dict[str, Any]: + return {"id": i, "store_code": i % 5 + 1, "status": "A" if i % 3 == 0 else "B", + "type": "web" if i % 2 else "shop", "level": i % 3 + 1, "pay_way": "cash" if i % 2 else "card", + "paid_at": None if i >= 39 else BASE + _dt.timedelta(days=i % 10), + "date": None if i >= 39 else (BASE + _dt.timedelta(days=i % 10)).date()} + + +ORDERS = [_order(i) for i in range(1, N_ORDERS + 1)] +VALID = [o for o in ORDERS if o["paid_at"] is not None] +IN_WINDOW = [o for o in VALID if o["paid_at"].date().isoformat() >= WINDOW_FROM] +EXPECT = { + "status": Counter(o["status"] for o in ORDERS), + "type": Counter(o["type"] for o in ORDERS), + "level": Counter(str(o["level"]) for o in ORDERS), + "pay_way": Counter(o["pay_way"] for o in ORDERS), + "paid_at": (min(o["paid_at"] for o in VALID), max(o["paid_at"] for o in VALID)), + "day_volume": Counter(o["paid_at"].date().isoformat() for o in IN_WINDOW), + "window_status": Counter(o["status"] for o in IN_WINDOW), +} + + +# ---------------------------------------------------------------- 各库的夹具 + + +class Target: + def __init__(self, key: str, kind: str, port: int) -> None: + self.key, self.kind, self.port = key, kind, port + self.oracle = kind == "oracle" + self.mysql = kind in ("mysql", "mariadb") + self.pg = kind == "postgresql" + + def admin_url(self) -> str: + if self.pg: + return f"postgresql+asyncpg://postgres:{PASSWORD}@127.0.0.1:{self.port}/it" + if self.mysql: + return f"mysql+aiomysql://root:{PASSWORD}@127.0.0.1:{self.port}/it" + return f"oracle+oracledb://it:{PASSWORD}@127.0.0.1:{self.port}/?service_name=FREEPDB1" + + def source(self) -> SimpleNamespace: + user = {"postgresql": "postgres", "mysql": "root", "mariadb": "root", "oracle": "it"}[self.kind] + return SimpleNamespace(id=f"it-{self.key}", name=f"it_{self.key}", kind=self.kind, host="127.0.0.1", + port=self.port, database="FREEPDB1" if self.oracle else "it", username=user, + password=encrypt(PASSWORD), options={}, readonly=True, description="", + schema_cache={}, origin="manual", enabled=True) + + +def _q(t: Target, name: str) -> str: + """建表时的列名：Oracle 上 TYPE、LEVEL、DATE 是保留字，按大写加引号建（库里就是大写的普通名字）。""" + if t.oracle and name in ("type", "level", "date"): + return f'"{name.upper()}"' + if t.mysql: + return f"`{name}`" + return f'"{name}"' if name in ("type", "level", "date") else name + + +def _ddl(t: Target) -> list[str]: + ts = "TIMESTAMP" if not t.mysql else "DATETIME" + intg = "NUMBER(10)" if t.oracle else "INTEGER" + vc = "VARCHAR2" if t.oracle else "VARCHAR" + cols = [f"id {intg} PRIMARY KEY", f"store_code {intg}", "status CHAR(2)", f"{_q(t, 'type')} {vc}(10)", + f"{_q(t, 'level')} {intg}", f"paid_at {ts}", f"{_q(t, 'date')} DATE"] + if t.mysql: + cols.append("pay_way ENUM('cash','card','coupon')") + out = [f"CREATE TABLE stores (code {vc}(10) PRIMARY KEY, name {vc}(20))", + f"CREATE TABLE orders ({', '.join(cols)})", + "CREATE INDEX ix_orders_paid_at ON orders (paid_at)"] + events = f"CREATE TABLE events (id {intg} PRIMARY KEY, kind {vc}(10))" + if t.pg: + events += " WITH (autovacuum_enabled = false)" + if t.mysql: + events += " STATS_PERSISTENT=1 STATS_AUTO_RECALC=0" + out.append(events) + if t.oracle: + out.append('CREATE TABLE "visits" ("id" NUMBER(10) PRIMARY KEY, "level" NUMBER(10))') + return out + + +def _paid(t: Target, o: dict[str, Any]) -> str: + if o["paid_at"] is not None: + stamp = o["paid_at"].strftime("%Y-%m-%d %H:%M:%S") + return f"TIMESTAMP '{stamp}'" if t.oracle else f"'{stamp}'" + # 无效日期：各家能存进去的那种 + if t.mysql: + return "'0000-00-00 00:00:00'" + if t.pg: + return "'infinity'" if o["id"] == 39 else "'-infinity'" + return "TIMESTAMP '9999-12-31 23:59:59'" if o["id"] == 39 else "TIMESTAMP '0001-01-01 00:00:00'" + + +def _rows(t: Target) -> list[str]: + out = [f"INSERT INTO stores (code, name) VALUES ('{c}', 'store {c}')" for c in ("1", "2", "3", "4", "5", "A1")] + for o in ORDERS: + day = f"DATE '{o['date'].isoformat()}'" if o["date"] is not None else "NULL" + cols = ["id", "store_code", "status", _q(t, "type"), _q(t, "level"), "paid_at", _q(t, "date")] + vals = [str(o["id"]), str(o["store_code"]), f"'{o['status']}'", f"'{o['type']}'", str(o["level"]), + _paid(t, o), day] + if t.mysql: + cols.append("pay_way") + vals.append(f"'{o['pay_way']}'") + out.append(f"INSERT INTO orders ({', '.join(cols)}) VALUES ({', '.join(vals)})") + if t.oracle: + out += ['INSERT INTO "visits" ("id", "level") VALUES (1, 1)', 'INSERT INTO "visits" ("id", "level") VALUES (2, 1)', + 'INSERT INTO "visits" ("id", "level") VALUES (3, 2)'] + return out + + +def _events(start: int, n: int) -> list[str]: + return [f"INSERT INTO events (id, kind) VALUES ({i}, '{'open' if i % 4 else 'close'}')" + for i in range(start, start + n)] + + +def _analyze(t: Target, table: str) -> list[str]: + if t.pg: + return [f"ANALYZE {table}"] + if t.mysql: + return [f"ANALYZE TABLE {table}"] + return [f"BEGIN DBMS_STATS.GATHER_TABLE_STATS(USER, '{table.upper()}'); END;"] + + +async def _wait(t: Target, deadline_s: float = 360) -> Any: + engine = create_async_engine(t.admin_url()) + started = time.monotonic() + while True: + try: + async with engine.connect() as conn: + await conn.execute(text("SELECT 1 FROM DUAL" if t.oracle else "SELECT 1")) + return engine + except Exception as e: # noqa: BLE001 - 容器还没起来：等 + if time.monotonic() - started > deadline_s: + raise RuntimeError(f"{t.key} 在 {deadline_s:.0f} 秒内没起来：{e}") from e + await asyncio.sleep(3) + + +async def _setup(t: Target) -> None: + engine = await _wait(t) + try: + async with engine.begin() as conn: + if t.mysql: + # 关掉严格模式才写得进零日期（老库里常见的情形） + await conn.execute(text("SET SESSION sql_mode = ''")) + for sql in _ddl(t) + _rows(t) + _events(1, N_EVENTS_OLD): + await conn.exec_driver_sql(sql) + async with engine.begin() as conn: + for sql in _analyze(t, "events") + _analyze(t, "orders") + _analyze(t, "stores"): + await conn.exec_driver_sql(sql) + if t.oracle: + await conn.exec_driver_sql("""BEGIN DBMS_STATS.GATHER_TABLE_STATS(USER, '"visits"'); END;""") + async with engine.begin() as conn: + if t.mysql: + await conn.execute(text("SET SESSION sql_mode = ''")) + for sql in _events(N_EVENTS_OLD + 1, N_EVENTS_NEW): + await conn.exec_driver_sql(sql) + finally: + await engine.dispose() + + +# ---------------------------------------------------------------- 剖析：目录换成内存里的 + + +async def _profile(source: Any, tables: list[str], notes: dict[str, dict[str, Any]]) -> dict[str, Any]: + cx = catalog_profile.make_context(source, SETTINGS, options={}) + captured: dict[str, Any] = {} + + async def db(fn: Any, *args: Any) -> Any: + if fn is catalog_profile._read_notes: + return notes.get(args[1], {}) + if fn is catalog_profile._record: + _sid, table, out, result, _actor, _rec = args + catalog_profile._settle_status(result) + captured[table] = (out, result) + return None + raise AssertionError(f"剖析要了没预料的库操作：{fn}") + + for table in tables: + await catalog_profile.profile_table(cx, table, db=db) + return captured + + +def _key(meta: dict[str, Any], name: str) -> str: + """表结构里这个名字的写法（Oracle 上普通名字是小写，TYPE 这类保留字是大写）。""" + return next(str(c["name"]) for c in meta["columns"] if str(c["name"]).lower() == name) + + +def _day(v: Any) -> str: + return str(v)[:10] + + +async def run_target(t: Target) -> None: + print(f"\n== {t.key}（{t.kind}，端口 {t.port}）") + await _setup(t) + source = t.source() + try: + cache = await introspect.introspect(source) + source.schema_cache = cache + tables = cache["tables"] + orders, events, stores = tables["orders"], tables["events"], tables["stores"] + types = {c["name"].lower(): c for c in orders["columns"]} + check(not cache.get("failed") and {"orders", "events", "stores"} <= set(tables), "探查读到三张夹具表") + check("paid_at" in [c.lower() for c in orders.get("index_first_columns") or []], + f"索引首列记进了表结构：{orders.get('index_first_columns')}") + if t.oracle: + check(stores["columns"][0]["type"] == "VARCHAR2(10 CHAR)", f"Oracle 文字类型写全了长度和单位：" + f"{stores['columns'][0]['type']}") + check(types["id"]["type"] == "NUMBER(10, 0)", f"Oracle 数字类型写全了精度：{types['id']['type']}") + check("visits" in tables and tables["visits"].get("quote_name") is True, + "Oracle 小写带引号的表记了 quote_name") + check(all(c.get("quote_name") is True for c in tables["visits"]["columns"]), + "Oracle 小写带引号的列记了 quote_name") + check(_key(orders, "level") == "LEVEL" and "quote_name" not in types["level"], + "Oracle 的保留字列 LEVEL 是大写的普通名字，不记 quote_name") + if t.mysql: + check(types["pay_way"]["type"].startswith("ENUM(") and + types["pay_way"].get("enum_values") == ["cash", "card", "coupon"], + f"MySQL ENUM 记下了取值列表：{types['pay_way']}") + if t.pg: + check(types["paid_at"]["type"] == "TIMESTAMP WITHOUT TIME ZONE", + f"PostgreSQL 时间类型写全了时区：{types['paid_at']['type']}") + + store_col, code_col = _key(orders, "store_code"), _key(stores, "code") + rel = catalog.relation_id("orders", [store_col], "stores", [code_col]) + notes = {"orders": {"relations": [{"id": rel, "columns": [store_col], "to_table": "stores", + "to_columns": [code_col], "cardinality": "many_to_one", "source": "name", + "status": "proposed"}]}} + names = ["orders", "events", "stores"] + (["visits"] if t.oracle else []) + got = await _profile(source, names, notes) + + logs = [q for _, r in got.values() for q in r.query_log] + bad = [f"{q['kind']} {q['target']}：{q['outcome']} {q['detail']}" for q in logs if q["outcome"] != "ok"] + check(not bad, f"剖析的 {len(logs)} 条查询零报错" + (f"：{bad[:3]}" if bad else "")) + + out, result = got["orders"] + check(result.status == "ok", f"orders 剖析完成：{result.status} {result.error or ''}") + [r] = out.relations or [None] + check(r is not None and r.coverage == 1.0 and r.sample == 5 and r.matched == 5 and r.holds, + f"整数子列对文字主键：取 5 个键值全对上、升为有确证（{r and r.note}）") + codes = {c.column.lower(): c for c in out.codes} + for col in ("status", "type", "level") + (("pay_way",) if t.mysql else ()): + c = codes.get(col) + check(c is not None and Counter(dict(c.values)) == EXPECT[col], + f"{col} 的取值分布和手算一致：{c and c.values}") + if t.oracle or t.pg: + check(codes["status"].values and all(not k.endswith(" ") for k, _ in codes["status"].values), + "CHAR(2) 码值去掉了补齐的空格") + if t.mysql: + check(codes["pay_way"].declared == ["coupon"] and "coupon" in codes["pay_way"].note, + "ENUM 列定义里没出现的取值也进了码值") + check(catalog_profile.MYSQL_CODES_COLLATION_NOTE in codes["status"].note, "MySQL 码值说明写了排序规则") + check(catalog_profile.MYSQL_COLLATION_NOTE in r.note, "MySQL 关系说明写了排序规则") + ranges = {d["column"].lower(): d for d in result.date_ranges} + paid = ranges.get("paid_at") + lo, hi = EXPECT["paid_at"] + check(paid is not None and paid["min"][:19].replace("T", " ") == str(lo) and + paid["max"][:19].replace("T", " ") == str(hi) and paid.get("invalid") == 2, + f"paid_at 的范围排除了 2 个无效日期：{paid}") + day = ranges.get("date") + check(day is not None and _day(day["min"]) == lo.date().isoformat() and _day(day["max"]) == + hi.date().isoformat() and not day.get("invalid"), f"date 列（保留字）的范围：{day}") + + out_e, result_e = got["events"] + size = result_e.row_estimate + d = SqlDialect(t.kind) + stats = await data_engine.run_query(source, d.stats_sql(events, "events")) + estimated = int(float(stats.rows[0][0] or 0)) if stats.rows else 0 + if estimated <= SETTINGS.max_scan_rows: + check(size is not None and size.rows is None and size.at_least == SETTINGS.max_scan_rows + 1, + f"统计信息做旧（说 {estimated} 行，其实 {N_EVENTS_OLD + N_EVENTS_NEW} 行）也认出了大表：{size}") + else: + # 有的库插入时自己就更新了行数估算，夹具做不旧：照实说，只核对按统计信息认出了大表 + print(f" · 统计信息没能做旧（库自己更新成了 {estimated} 行），只核对按统计信息认出大表") + check(size is not None and size.method == "stats" and (size.rows or 0) > SETTINGS.max_scan_rows, + f"按统计信息认出了大表：{size}") + kind_codes = next((c for c in out_e.codes if c.column.lower() == "kind"), None) + check(kind_codes is not None and not kind_codes.complete and "按存储顺序取前" in kind_codes.note, + f"大表的码值按存储顺序的前若干行，不标已列全：{kind_codes and kind_codes.note}") + + if t.oracle: + out_v, result_v = got["visits"] + lv = next((c for c in out_v.codes if c.column == "level"), None) + check(result_v.status == "ok" and lv is not None and Counter(dict(lv.values)) == Counter({"1": 2, "2": 1}), + f"小写带引号的表和列照样剖析：{result_v.status} {lv and lv.values}") + vstats = await data_engine.run_query(source, d.stats_sql(tables["visits"], "visits")) + check(bool(vstats.rows) and vstats.rows[0][0] is not None, f"小写表的统计信息按原名查得到：{vstats.rows}") + + # ---- 取值观察的新方法 + tbl = d.table(orders, "orders") + paid_col = d.column(orders, _key(orders, "paid_at")) + res = await data_engine.run_query(source, d.day_volume_sql(tbl, paid_col, WINDOW_FROM)) + got_days = Counter({_day(row[0]): int(row[1]) for row in res.rows}) + check(got_days == EXPECT["day_volume"], f"按日行数和手算一致（无效日期不算进窗口）：{dict(got_days)}") + status_col = d.column(orders, _key(orders, "status")) + res = await data_engine.run_query(source, d.window_value_counts_sql(tbl, status_col, paid_col, WINDOW_FROM, 21)) + got_window = Counter({str(row[0]).rstrip(" "): int(row[1]) for row in res.rows}) + check(got_window == EXPECT["window_status"], f"时间窗口里的取值分布和手算一致：{dict(got_window)}") + sampled = d.sample_value_counts_sql(d.table(events, "events"), d.column(events, _key(events, "kind")), 50, 21) + if t.pg or t.oracle: + res = await data_engine.run_query(source, sampled) + total = sum(int(row[1]) for row in res.rows) + check(total <= N_EVENTS_OLD + N_EVENTS_NEW, f"块采样跑得通（取到 {total} 行）") + else: + check(sampled is None, "MySQL / MariaDB 没有块采样，照实返回 None") + finally: + await data_engine.engines.invalidate(source.id) + + +async def main() -> int: + only = set((os.environ.get("DIALECT_IT_ONLY") or "pg,mysql,mariadb,oracle").split(",")) + plan = [("pg", "postgresql", "DIALECT_IT_PG_PORT"), ("mysql", "mysql", "DIALECT_IT_MYSQL_PORT"), + ("mariadb", "mariadb", "DIALECT_IT_MARIADB_PORT"), ("oracle", "oracle", "DIALECT_IT_ORACLE_PORT")] + for key, kind, env in plan: + if key not in only: + continue + port = os.environ.get(env) + if not port: + failures.append(f"{key}：没有拿到容器端口（{env}）") + continue + try: + await run_target(Target(key, kind, int(port))) + except Exception as e: # noqa: BLE001 - 一种库出错不挡别的库，照实记下 + print(f" ✗ {key} 中途出错：{type(e).__name__}: {e}") + failures.append(f"{key}：{type(e).__name__}: {e}") + await data_engine.engines.close() + print("\n" + ("全部通过" if not failures else f"{len(failures)} 项不通过：\n- " + "\n- ".join(failures))) + return 1 if failures else 0 + + +if __name__ == "__main__": + sys.exit(asyncio.run(main()))