diff --git a/backend/app/data/catalog.py b/backend/app/data/catalog.py index 678f249b..1723aea3 100644 --- a/backend/app/data/catalog.py +++ b/backend/app/data/catalog.py @@ -567,18 +567,48 @@ def _undated(note: str) -> str: return _PROFILE_NOTE_DAY.sub("数据剖析", note, count=1) +#: 剖析备注改过的说法：(升级前的写法, 现在的写法)。大表看的是按存储顺序的前 N 行（多半是最老的数据）、关系取的是 +#: 数据库先给出的那几个键值，都不是随机抽样，说法改成照实的（data/catalog_profile.py 的 _relation_note、码值的 +#: 「统计范围」、FILTER_SAMPLED）。只换这几处固定的句式，换完和现在的写法一字不差才算同一句 +_LEGACY_PROFILE_WORDING = ( + (re.compile(r"抽样覆盖率 (?=\S+?，可能不是这条关系)"), "覆盖率 "), + (re.compile(r"子表抽样 (?=\d+ 个不同键值)"), "子表取了 "), + (re.compile(r"（只看前 (?=\d+ 行）)"), "（按存储顺序只看前 "), + (re.compile(r"按前 (?=[\d,]+ 行统计)"), "按存储顺序取前 "), +) +#: 起草时核对有效行过滤写进备注的那句（catalog_profile.FILTER_SAMPLED / FILTER_SAMPLED_EDGE）：接在模型给的依据后面， +#: 整段不以「数据剖析：」开头，只换这一句 +_LEGACY_FILTER_WORDING = re.compile(r"剖析核对：按前 (?=[\d,]+ 行统计)") + + +def _current_wording(note: str) -> str: + """剖析备注里升级前的说法换成现在的写法（_LEGACY_PROFILE_WORDING）。只用来比较，不改写目录里存着的备注。""" + for pattern, now in _LEGACY_PROFILE_WORDING: + note = pattern.sub(now, note) + return note + + def same_profile_note(a: Any, b: Any) -> bool: """两段备注是否相同；数据剖析写的备注只差开头的日期（升级前每次都写当天的日期），或者一段带日期、一段是不带 - 日期的新写法，也算相同。 + 日期的新写法，或者只差改过的几处说法（_LEGACY_PROFILE_WORDING：「抽样覆盖率」→「覆盖率」之类），也算相同。 隔天重跑、数据没变，结论一字不差、只有日期变了：要是这也算「变了」，目录就升一个版本，发布时记下的目录版本 - 对不上，之后每次正式运行都提醒「目录有变化」。升级后第一次重新剖析也是一样：结论没变，只是新写法不带日期。 + 对不上，之后每次正式运行都提醒「目录有变化」。升级后第一次重新剖析也是一样：结论没变，只是新写法不带日期、 + 措辞改了。算相同时保留旧的那段（目录不动），旧说法要等结论真的变了、重写备注时才换掉。 + + 起草写的备注（有效行过滤）里接着的剖析核对那句同理，只认那一句的旧说法（_LEGACY_FILTER_WORDING）。 """ if a == b: return True - if not (is_profile_note(a) and is_profile_note(b)): + if not (isinstance(a, str) and isinstance(b, str)): return False - return _undated(a) == _undated(b) + profile_a, profile_b = is_profile_note(a), is_profile_note(b) + if profile_a and profile_b: + return _current_wording(_undated(a)) == _current_wording(_undated(b)) + if profile_a or profile_b: + return False + return (_LEGACY_FILTER_WORDING.sub("剖析核对：按存储顺序取前 ", a) + == _LEGACY_FILTER_WORDING.sub("剖析核对：按存储顺序取前 ", b)) def _same_core(a: Mapping[str, Any], b: Mapping[str, Any]) -> bool: diff --git a/backend/app/data/catalog_profile.py b/backend/app/data/catalog_profile.py index cc9b1062..b484ab54 100644 --- a/backend/app/data/catalog_profile.py +++ b/backend/app/data/catalog_profile.py @@ -43,7 +43,7 @@ from collections.abc import Awaitable, Callable from dataclasses import asdict, dataclass, field from datetime import date as _date, datetime, timezone from decimal import Decimal -from typing import Any +from typing import Any, NamedTuple from sqlalchemy import select, update from sqlalchemy.dialects import mysql, oracle, postgresql, sqlite @@ -250,30 +250,42 @@ def is_fixed_char(type_name: str | None) -> bool: return bool(type_name) and bool(_FIXED_CHAR.match(str(type_name))) +class KeyLiterals(NamedTuple): + """key_literals 的结果。literals：去父表里对的 IN 列表字面量（去重、保持顺序）；unmatchable：按类型就对不上父键、 + 算作没对上的键值个数（去重后）；unsafe：值本身不能安全拼进 SQL、不计入核对的个数。""" + literals: list[str] + unmatchable: int + unsafe: int + + def key_literals(values: list[Any], kind: str | None, *, parent: str | None, - child_type: str | None) -> tuple[list[str], int]: - """从子表取回的键值 → 去父表里对的 IN 列表字面量（去重、保持顺序）：(字面量, 因为两端类型不同没拼的个数)。 + child_type: str | None) -> KeyLiterals: + """从子表取回的键值 → 去父表里对的 IN 列表字面量（KeyLiterals）。 字面量按**被指向列**的类型族写（catalog.column_family），不是按子表取回的值： - 父列是文字、子列是数：写成 '123'。PostgreSQL 上 varchar 列 IN (123) 直接报「operator does not exist」； - - 父列是数、子列是文字：只收整数写法的值（'0012' 也照 12 写），别的（'A1'）计入不同类型、不拼。Oracle 上 - 数字列 IN ('A1') 会把字面量转成数报 ORA-01722，MySQL 会把 'A1' 转成 0 去比、对上不该对上的行； - - 父列是时间：键不核对（各家时间字面量写法不同、精度不同），全部计入不同类型； + - 父列是数、子列是文字：只拼整数写法的值（'0012' 也照 12 写）。别的（'A1'、'WALKIN'）不拼——Oracle 上数字列 + IN ('A1') 会把字面量转成数报 ORA-01722，MySQL 会把 'A1' 转成 0 去比、对上不该对上的行——但它们**照样是子表 + 的键值**：整数主键里永远没有 'WALKIN'，所以算作没对上，计入 unmatchable，由调用方留在覆盖率的分母里。要是 + 从分母里拿掉，散客代号占九成的一列只剩 1 个整数写法的键对上，就成了「覆盖率 100%」； + - 父列是时间：不在这里处理，_check_relation 先整条跳过（各家时间字面量写法不同、精度不同，键不核对）； - 父列类型认不出：照子表取回的值的类型写（sql_literal，以前的做法）。 子表是定长文字（CHAR(n)）时先去掉尾随空格：PG、Oracle 取回的 'A ' 拿去和 VARCHAR 父列比就对不上，而按 CHAR 的 - 语义尾随空格本来就不算。不安全的值（反斜杠、「:名字」、控制字符、太长）照 sql_literal 的规矩跳过，不算类型不同。 + 语义尾随空格本来就不算。不安全的值（反斜杠、「:名字」、控制字符、太长）照 sql_literal 的规矩跳过，记进 unsafe + （调用方另记「类型不支持核对，未计入」）。两个不同的取值写成同一个字面量（'0012' 和 '12'）只算一个，不算 unsafe。 """ + if parent == "time": + raise ValueError("被指向列是日期时间的关系不核对键值（见 _check_relation）") fixed = is_fixed_char(child_type) out: list[str] = [] - mismatched = 0 + unmatchable: set[str] = set() + unsafe = 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): + unsafe += 1 continue if isinstance(value, (int, float, Decimal)): lit = sql_literal(value, "number") @@ -284,16 +296,18 @@ def key_literals(values: list[Any], kind: str | None, *, parent: str | None, if isinstance(value, str): text = value.strip() if not _INT_TEXT.fullmatch(text): - mismatched += 1 + unmatchable.add(text) continue lit = str(int(text)) else: lit = sql_literal(value, "number") else: lit = sql_literal(value, kind) - if lit is not None: + if lit is None: + unsafe += 1 + else: out.append(lit) - return list(dict.fromkeys(out)), mismatched + return KeyLiterals(list(dict.fromkeys(out)), len(unmatchable), unsafe) _DIALECT_MODULES = {"sqlite": sqlite, "postgres": postgresql, "postgresql": postgresql, "mysql": mysql, @@ -1152,12 +1166,13 @@ def _collation_note(dialect: SqlDialect, family: str | None, note: str) -> str | 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, mismatched: int = 0, collation: str | None = None) -> str: + holds: bool, unmatchable: int = 0, collation: str | None = None) -> str: """关系的剖析备注：取了多少个键值、看了多少行、覆盖率、父键和子键唯一的依据。人据此判断这条结论有多可靠。 不写日期：什么时候剖析的看这一项的 updated_at（界面按本地时间显示）。 不说「抽样」：DISTINCT … LIMIT 取的是数据库先给出的那几个值，大表上是按存储顺序的前若干行里的值，不是随机抽的 - （多半是最老的数据）。mismatched：和被指向列类型不同、没拼进核对的键值个数（key_literals）。""" + （多半是最老的数据）。unmatchable：被指向列是数字、子表取到的却不是整数写法的键值个数（key_literals）——已经算在 + sample 里、算作没对上，这里说明为什么没去父表里对。""" parts: list[str] = [] low = coverage < catalog.PROFILE_VERIFY_COVERAGE if low: @@ -1167,8 +1182,8 @@ def _relation_note(*, sample: int, matched: int, coverage: float, scan_cap: int 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 unmatchable: + parts.append(f"其中 {unmatchable} 个不是整数写法，被指向列是数字，算作没对上") if collation: parts.append(collation) unique, how = parent @@ -1214,6 +1229,11 @@ async def _check_relation(cx: _TableCx, table: str, rel: dict[str, Any], out: _T return d, s = cx.dialect, cx.settings + child_col, parent_col = _column_meta(meta, col), _column_meta(pmeta, pcol) + parent_family = catalog.column_family(parent_col) + if parent_family == "time": + # 时间键不核对：各家时间字面量写法、精度不同，拼出来的 IN 列表对不对得上说不准。不发子表的取值查询 + raise _Skip("unsupported", f"{parent}.{pcol} 是日期时间，各库的时间写法和精度不同，不核对这种关系的覆盖率") 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) @@ -1221,24 +1241,20 @@ async def _check_relation(cx: _TableCx, table: str, rel: dict[str, Any], out: _T 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] - 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")) + literals, unmatchable, unsafe = 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: + if not literals and not unmatchable: raise _Skip("unsupported", f"{table}.{col} 的取值类型不支持核对（只核对整数和文字）") + # 分母是子表取到的全部键值：不是整数写法、按类型就对不上整数父键的（unmatchable）也在里面，算作没对上 + sample = len(literals) + unmatchable matched = 0 for start in range(0, len(literals), 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) + coverage = round(matched / sample, 4) parent_unique = await _unique(cx, parent, pmeta, pcol, result) child_unique = await _unique(cx, table, meta, col, result) if parent_unique[0] else None @@ -1246,14 +1262,13 @@ 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=max(unsafe, 0), parent=parent_unique, child=child_unique, holds=holds, - mismatched=mismatched, + note = _relation_note(sample=sample, matched=matched, coverage=coverage, scan_cap=scan_cap, + skipped_values=unsafe, parent=parent_unique, child=child_unique, holds=holds, + unmatchable=unmatchable, 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), + cardinality=cardinality, holds=holds, sample=sample, matched=matched, note=note, cardinality_checked=checked)) @@ -1468,7 +1483,7 @@ def _relation_finding(o: _RelationOutcome, rel: dict[str, Any], *, status: str, elif status == "verified": summary = f"覆盖率 {_percent(o.coverage)}，" + ("仍为有确证" if previous == "verified" else "升为有确证") elif o.coverage < catalog.PROFILE_VERIFY_COVERAGE: - summary = f"抽样覆盖率 {_percent(o.coverage)}，保持推断" + summary = f"覆盖率 {_percent(o.coverage)}，保持推断" else: summary = "被指向列未核实唯一，保持推断" return {"kind": "relation", "path": f"relations.{o.rid}", "target": o.target, diff --git a/backend/tests/test_catalog_profile.py b/backend/tests/test_catalog_profile.py index 2a04fe76..0a1ef009 100644 --- a/backend/tests/test_catalog_profile.py +++ b/backend/tests/test_catalog_profile.py @@ -23,7 +23,7 @@ from httpx import ASGITransport, AsyncClient from sqlalchemy import event from sqlalchemy.engine import Engine -from app.data import catalog_profile, catalog_profile_jobs +from app.data import catalog, catalog_profile, catalog_profile_jobs from app.data import engine as data_engine from app.data.catalog_profile import PROFILE_OPTION, ProfileSettings from app.db.base import SessionLocal @@ -327,6 +327,39 @@ async def test_wrong_name_relation_stays_proposed_with_coverage_in_note(client, assert "business_date" not in (await _notes(client, sid, "parking_records"))["notes"] +async def test_text_keys_that_cannot_be_integers_count_as_unmatched(client, make_source, tmp_path): + """模型起草的关系：orders.customer_ref（文字）指向 customers.id（整数主键），可取值大多是散客代号。不是整数写法的 + 文字永远对不上整数主键，算作没对上、留在分母里——不能从分母里拿掉，把 1 个对上的键说成覆盖率 100% 升为有确证。""" + path = str(tmp_path / "walkin.db") + db = sqlite3.connect(path) + try: + db.executescript("CREATE TABLE customers (id INTEGER PRIMARY KEY, name TEXT NOT NULL);" + "CREATE TABLE orders (id INTEGER PRIMARY KEY, customer_ref TEXT, amount INTEGER);") + db.executemany("INSERT INTO customers VALUES (?, ?)", [(i, f"会员{i}") for i in (1, 2, 3)]) + refs = ["WALKIN"] * 90 + ["1"] * 10 + [f"GUEST{i}" for i in range(30)] + db.executemany("INSERT INTO orders VALUES (?, ?, ?)", [(i, ref, 10) for i, ref in enumerate(refs, 1)]) + db.commit() + finally: + db.close() + sid = await make_source(path, options=_on()) + rel_id = catalog.relation_id("orders", ["customer_ref"], "customers", ["id"]) + async with SessionLocal() as session: + await catalog.write_entry(session, sid, "orders", {"relations": [ + {"id": rel_id, "columns": ["customer_ref"], "to_table": "customers", "to_columns": ["id"], + "cardinality": "many_to_one", "coverage": None, "source": "llm", "status": "proposed"}]}, + if_version=0, actor=None) + + body = await _profile(client, sid, ["orders"]) + rel = _relation((await _notes(client, sid, "orders"))["notes"], "customer_ref") + assert (rel["source"], rel["status"]) == ("profile", "proposed") + assert rel["coverage"] == round(1 / 32, 4) + assert "覆盖率 3.1%" in rel["note"] and "可能不是这条关系" in rel["note"] + assert "子表取了 32 个不同键值，父表对上 1 个" in rel["note"] + assert "31 个不是整数写法" in rel["note"] and "未计入" not in rel["note"] + finding = _table(body, "orders")["findings"][0] + assert (finding["status"], finding["sample"], finding["matched"]) == ("proposed", 32, 1) + + async def test_confirmed_relation_only_gets_coverage_and_cardinality(client, make_source, scenic_db): sid = await make_source(scenic_db, options=_on(), draft=["visits"]) detail = await _notes(client, sid, "visits") diff --git a/backend/tests/test_catalog_semantics.py b/backend/tests/test_catalog_semantics.py index 5501a4d8..d15e0332 100644 --- a/backend/tests/test_catalog_semantics.py +++ b/backend/tests/test_catalog_semantics.py @@ -14,6 +14,7 @@ """ from __future__ import annotations +import re import sqlite3 import pytest @@ -408,6 +409,95 @@ async def test_rerun_on_another_day_with_same_data_keeps_the_version(client, mak assert codes["note"].startswith("数据剖析：") and "2026-10-03" not in codes["note"] +#: 升级前剖析写的说法 → 现在的说法（下一轮 C1：大表看的是按存储顺序的前 N 行，不是抽样）。测试里反过来用：把新写法 +#: 改回旧写法，模拟升级前写进目录的备注 +_WORDING = (("覆盖率 ", "抽样覆盖率 ", "，可能不是这条关系"), ("子表取了 ", "子表抽样 ", None), + ("（按存储顺序只看前 ", "（只看前 ", None), ("按存储顺序取前 ", "按前 ", None)) + + +def _legacy(note: str) -> str: + out = note + for new, old, tail in _WORDING: + if tail: + out = re.sub(re.escape(new) + r"(\S+?)" + re.escape(tail), lambda m: old + m.group(1) + tail, out) + else: + out = out.replace(new, old) + return out + + +def _profile_notes(notes: dict) -> list[dict]: + """目录里剖析写了备注的项（关系、码值、业务日期）。""" + found = [r for r in notes.get("relations") or [] if catalog.is_profile_note(r.get("note"))] + for column in (notes.get("columns") or {}).values(): + found += [item for item in column.values() if isinstance(item, dict) + and catalog.is_profile_note(item.get("note"))] + bdate = notes.get("business_date") + return found + ([bdate] if isinstance(bdate, dict) and catalog.is_profile_note(bdate.get("note")) else []) + + +def test_profile_notes_in_the_old_wording_are_the_same_note(): + """改过的四处说法：旧写法和新写法一字之外都一样才算同一句；数变了、或者不是剖析写的备注，照旧算不同。""" + old = "数据剖析：抽样覆盖率 15%，可能不是这条关系；子表抽样 20 个不同键值（只看前 100 行），父表对上 3 个；被指向列是主键。" + new = "数据剖析：覆盖率 15%，可能不是这条关系；子表取了 20 个不同键值（按存储顺序只看前 100 行），父表对上 3 个；被指向列是主键。" + assert catalog.same_profile_note(old, new) and catalog.same_profile_note(new, old) + assert catalog.same_profile_note(old.replace("数据剖析：", "数据剖析（2026-10-01）："), new) + assert not catalog.same_profile_note(old, new.replace("对上 3 个", "对上 4 个")) + codes_old = "数据剖析（2026-10-01）：按前 10 行统计，10 行非空值共 2 个取值：1（90%）、0（10%）。" + codes_new = "数据剖析：按存储顺序取前 10 行统计，10 行非空值共 2 个取值：1（90%）、0（10%）。" + assert catalog.same_profile_note(codes_old, codes_new) + assert not catalog.same_profile_note(codes_old, codes_new.replace("前 10 行", "前 20 行")) + # 起草写的有效行过滤备注：只认接在后面的剖析核对那句 + filter_old = "依据：status 为 1 才是有效记录。剖析核对：按前 100,000 行统计，排除 3% 的行" + filter_new = "依据：status 为 1 才是有效记录。剖析核对：按存储顺序取前 100,000 行统计，排除 3% 的行" + assert catalog.same_profile_note(filter_old, filter_new) + assert not catalog.same_profile_note(filter_old, filter_new.replace("3%", "4%")) + # 不是剖析写的备注：「抽样覆盖率」是人写的原话，不换 + assert not catalog.same_profile_note("人工备注：抽样覆盖率 15%，可能不是这条关系", + "人工备注：覆盖率 15%，可能不是这条关系") + assert not catalog.same_profile_note(new, "覆盖率 15%，可能不是这条关系") + + +async def test_rerun_after_upgrade_keeps_the_version_when_only_the_wording_changed(client, make_source, tmp_path): + """升级前写进目录的剖析备注用的是旧说法（「抽样覆盖率」「子表抽样」「（只看前 N 行）」「按前 N 行统计」）。数据 + 没变、升级后第一次重新剖析：结论一字不差、只是措辞改了，不算变化——否则每张有剖析结论的表都升一个版本，发布时 + 记下的目录版本对不上，之后每次正式运行都提醒「目录有变化」。""" + path = scenic.build(tmp_path / "wording.db") + db = sqlite3.connect(path) + try: + db.execute("CREATE TABLE lots (id INTEGER PRIMARY KEY, name TEXT NOT NULL)") + db.executemany("INSERT INTO lots VALUES (?, ?)", [(i, f"{i} 号停车区") for i in (1, 2, 3)]) + db.executemany("INSERT INTO parking_records VALUES (?, ?, ?, ?, ?, ?)", + [(i, 1 + i % 20, None, "2026-08-01 09:00:00", None, 10.0) for i in range(1, 101)]) + db.commit() + finally: + db.close() + tables = ["visits", "parking_records"] + whole = await make_source(path, options=_on(), draft=tables) + capped = await make_source(path, options=_on(max_scan_rows=0, sample_size=10), draft=tables) + seen: set[str] = set() + for sid in (whole, capped): + await _profile(client, sid, tables) + for table in tables: + detail = await _notes(client, sid, table) + notes = detail["notes"] + for item in _profile_notes(notes): + old = _legacy(item["note"]) + seen |= {o for _, o, _ in _WORDING if o in old} + item["note"] = old + async with SessionLocal() as session: + await catalog.write_entry(session, sid, table, notes, if_version=detail["version"], actor=None) + assert seen == {o for _, o, _ in _WORDING}, "四处旧说法都写进了目录" + + for sid in (whole, capped): + before = {t: await _notes(client, sid, t) for t in tables} + body = await _profile(client, sid, tables) + for table in tables: + done = _table(body, table) + assert (done["added"], done["updated"], done["removed"]) == (0, 0, 0), (table, done) + assert done["version"] == before[table]["version"] + assert (await _notes(client, sid, table))["notes"] == before[table]["notes"] + + def test_merge_ignores_a_profile_note_that_only_changed_its_date(): old = {"columns": {"status": {"codes": catalog.make_item( {"1": "", "0": ""}, "profile", "proposed", note="数据剖析（2026-10-01）：统计全表，1500 行非空值共 2 个取值。", diff --git a/backend/tests/test_profile_dialect_golden.py b/backend/tests/test_profile_dialect_golden.py index 2eb0615c..7e7c6321 100644 --- a/backend/tests/test_profile_dialect_golden.py +++ b/backend/tests/test_profile_dialect_golden.py @@ -53,7 +53,7 @@ def _queries(kind: str) -> list[tuple[str, str | None]]: 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") + keys = key_literals([12, 7], "number", parent="text", child_type="INTEGER").literals out: list[tuple[str, str | None]] = [ ("stats", d.stats_sql(META["orders"], "orders")), ("stats_quoted", d.stats_sql(META["visits"], "visits")), @@ -158,7 +158,7 @@ def _in_families(sql: str, kind: str, schema: dict[str, dict[str, str]]) -> list 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) + literals = key_literals(child_values, None, parent=catalog.column_family(parent), child_type=child_type).literals assert literals sql = d.match_count_sql("stores", d.quote("code"), literals[:IN_CHUNK]) found = _in_families(sql, kind, {"stores": {"code": parent_type}}) diff --git a/backend/tests/test_profile_dialect_units.py b/backend/tests/test_profile_dialect_units.py index db01b746..ff04b0c7 100644 --- a/backend/tests/test_profile_dialect_units.py +++ b/backend/tests/test_profile_dialect_units.py @@ -135,22 +135,94 @@ def test_index_first_columns_are_recorded(): 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([12, 7], "number", parent="text", child_type="INTEGER") == ( + ["'12'", "'7'"], 0, 0) + # 父列是数、子列是文字：只拼整数写法；'A1' 按类型就对不上整数父键，记进 unmatchable（调用方算作没对上） + assert catalog_profile.key_literals(["12", "A1", "-3", "A1 "], "text", parent="number", + child_type="VARCHAR(10)") == (["12", "-3"], 1, 0) + # 父列是时间的关系整条不核对（_check_relation 先跳过），这里不收 + with pytest.raises(ValueError): + catalog_profile.key_literals([1, 2], "number", parent="time", child_type="INTEGER") # 父列类型不知道：照子表取回的值的类型写（以前的做法） - assert catalog_profile.key_literals([3, "x"], None, parent=None, child_type=None) == (["3", "'x'"], 0) + assert catalog_profile.key_literals([3, "x"], None, parent=None, child_type=None) == (["3", "'x'"], 0, 0) + + +def test_key_literals_count_unsafe_values_but_not_duplicate_spellings(): + """不能安全拼进 SQL 的值记进 unsafe（备注写「类型不支持核对，未计入」）；'0012' 和 '12' 写成同一个字面量，只算一个， + 不能被当成「不支持核对」。""" + found = catalog_profile.key_literals(["0012", "12", "a\\b", True], "text", parent="text", child_type="TEXT") + assert found == (["'0012'", "'12'"], 0, 2) + found = catalog_profile.key_literals(["0012", "12"], "text", parent="number", child_type="TEXT") + assert found == (["12"], 0, 0) + + +_REL = {"id": "r1", "columns": ["customer_ref"], "to_table": "customers", "to_columns": ["id"], "source": "llm", + "status": "proposed"} + + +def _relation_cx(values: list[Any], parent_type: str, matched: int): + """orders.customer_ref（文字）→ customers.id（parent_type）的关系核对用的执行状态：取值查询回 values，父表的核对 + 查询回 matched，唯一性计数回「有重复」。seen 记下发出的每条 SQL。""" + settings = ProfileSettings(enabled=True) + source = type("S", (), {"id": "s1", "name": "shop", "kind": "postgresql", "options": {}})() + tables = {"orders": {"columns": [{"name": "id", "type": "INTEGER"}, {"name": "customer_ref", "type": "TEXT"}], + "primary_key": ["id"]}, + "customers": {"columns": [{"name": "id", "type": parent_type}], "primary_key": ["id"]}} + job = catalog_profile.ProfileContext(source=source, settings=settings, dialect=SqlDialect("postgresql"), + runner=None, tables=tables, masked=set()) + job.sizes.update({"orders": TableSize(130, "count"), "customers": TableSize(3, "count")}) + cx = catalog_profile._TableCx(job, "orders", catalog_profile.TableResult(table="orders"), + catalog_profile.TableHooks()) + seen: list[str] = [] + + async def run(sql: str, **_kw: Any) -> QueryResult: + seen.append(sql) + if "SELECT DISTINCT" in sql: + rows = [[v] for v in values] + return QueryResult(columns=["k"], rows=rows, row_count=len(rows), truncated=False, elapsed_ms=1, sql=sql, + column_types={"k": "text"}) + if " IN (" in sql: + return _result(matched) + return QueryResult(columns=["a", "b"], rows=[[130, len(values)]], row_count=1, truncated=False, elapsed_ms=1, + sql=sql) + cx.run = run # type: ignore[method-assign] + return cx, seen + + +@pytest.mark.parametrize("values, parent_type, coverage, sample, matched", [ + # 九成是散客代号、只有 1 个整数写法对得上：以前分母里拿掉了对不上类型的，测成 100% + (["WALKIN", "1"] + [f"GUEST{i}" for i in range(30)], "INTEGER", round(1 / 32, 4), 32, 1), + # 全都不是整数写法：一条 IN 都不用发，覆盖率 0%，照样是「可能不是这条关系」，不是「无法核对」 + (["WALKIN", "GUEST1"], "INTEGER", 0.0, 2, 0), + (["1", "2"], "INTEGER", 1.0, 2, 2), +]) +def test_relation_coverage_counts_type_mismatches_as_unmatched(values, parent_type, coverage, sample, matched): + cx, seen = _relation_cx(values, parent_type, matched) + out, result = catalog_profile._TableOutcome(), catalog_profile.TableResult(table="orders") + asyncio.run(catalog_profile._check_relation(cx, "orders", _REL, out, result)) + [rel] = out.relations + assert (rel.coverage, rel.sample, rel.matched, rel.holds) == (coverage, sample, matched, coverage == 1.0) + if coverage < 1: + assert "可能不是这条关系" in rel.note and "未计入" not in rel.note + if sample > matched and matched == 0: + assert not any("IN (" in sql for sql in seen), "没有整数写法的键值：不发父表的核对查询" + + +def test_time_keys_are_not_checked_and_send_no_query(): + cx, seen = _relation_cx(["2026-09-01"], "DATE", 0) + out, result = catalog_profile._TableOutcome(), catalog_profile.TableResult(table="orders") + with pytest.raises(catalog_profile._Skip) as skipped: + asyncio.run(catalog_profile._check_relation(cx, "orders", _REL, out, result)) + assert skipped.value.reason == "unsupported" and "日期时间" in str(skipped.value) + assert "两端类型不同" not in str(skipped.value) + assert seen == [], "时间键不核对，一条查询都不发" 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)") + lits = catalog_profile.key_literals(["A ", "BC"], "text", parent="text", child_type="CHAR(2)").literals assert lits == ["'A'", "'BC'"] - lits, _ = catalog_profile.key_literals(["A "], "text", parent="text", child_type="VARCHAR(2)") + lits = catalog_profile.key_literals(["A "], "text", parent="text", child_type="VARCHAR(2)").literals assert lits == ["'A '"] @@ -283,7 +355,7 @@ def test_sample_percent_must_be_between_0_and_100(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, + unmatchable=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 行统计") @@ -291,6 +363,6 @@ def test_notes_say_storage_order_not_sampling(): 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, + unmatchable=0, parent=(True, "主键"), child=(False, "count"), holds=True, collation=catalog_profile.MYSQL_COLLATION_NOTE) assert "排序规则" in note and "不分大小写" in note