diff --git a/backend/app/data/catalog_profile.py b/backend/app/data/catalog_profile.py index 7198e95d..ed63b2ed 100644 --- a/backend/app/data/catalog_profile.py +++ b/backend/app/data/catalog_profile.py @@ -1509,7 +1509,8 @@ def _relation_finding(o: _RelationOutcome, rel: dict[str, Any], *, status: str, return {"kind": "relation", "path": f"relations.{o.rid}", "target": o.target, "columns": list(rel.get("columns") or []), "to_table": rel.get("to_table"), "to_columns": list(rel.get("to_columns") or []), "status": status, "confirmed": confirmed, - "previous_status": previous, "coverage": o.coverage, "cardinality": o.cardinality, + "previous_status": previous, "previous_coverage": rel.get("coverage"), "coverage": o.coverage, + "cardinality": o.cardinality, "cardinality_checked": o.cardinality_checked, "sample": o.sample, "matched": o.matched, "summary": summary} diff --git a/backend/app/data/catalog_profile_jobs.py b/backend/app/data/catalog_profile_jobs.py index df7cf3df..e24d2c19 100644 --- a/backend/app/data/catalog_profile_jobs.py +++ b/backend/app/data/catalog_profile_jobs.py @@ -159,13 +159,21 @@ async def task_rows(session: AsyncSession, job_id: str) -> list[Any]: )).all()) +#: 取值观察的四类检查在逐表进度里合成一类（observe） +_OBSERVE_GROUP = frozenset({"rows", "codes_observe", "freshness", "daily_volume"}) + + def _by_kind(checks: list[Any]) -> dict[str, list[int]]: - """检查项按种类的进度：{relation: [结束了的, 总数], codes: […], date: […]}。结束了的含跳过的。""" - out = {"relation": [0, 0], "codes": [0, 0], "date": [0, 0]} + """检查项按种类的进度：{relation: [结束了的, 总数], codes: […], date: […], observe: […]}。结束了的含跳过的；取值观察的 + 四类合成 observe 一类。""" + out = {"relation": [0, 0], "codes": [0, 0], "date": [0, 0], "observe": [0, 0]} for check in checks: - if not isinstance(check, dict) or check.get("kind") not in out: + if not isinstance(check, dict): continue - pair = out[check["kind"]] + kind = "observe" if check.get("kind") in _OBSERVE_GROUP else check.get("kind") + if kind not in out: + continue + pair = out[kind] pair[1] += 1 if check.get("state") in ("done", "skipped"): pair[0] += 1 @@ -1111,15 +1119,19 @@ class _ProfileRun(runtime.JobRun): if values := self._counters(): await session.execute(update(CatalogProfileJob).where(CatalogProfileJob.id == self.job_id).values(**values) .execution_options(synchronize_session=False)) - if result.observations and self.cx is not None: - # 取值观察：观察记录、比出的取值变化、撤销「已列全」都和这张表的结果、目录在同一个事务里 + if self.cx is not None and (result.observations or result.findings): + # 取值观察：观察记录、比出的取值变化、撤销「已列全」，以及关联覆盖率的下降，都和这张表的结果、目录在同一个 + # 事务里 from app.data import catalog_profile_observe from app.data.source_changes import Detected detected = Detected(job_kind="profile", job_id=self.job_id, task_seq=seq, trigger=self._trigger["trigger"], actor=self.actor, fire_at=self._trigger["fire_at"]) - await catalog_profile_observe.save(session, self.cx.source, result.table, result, job_id=self.job_id, - seq=seq, detected=detected) + if result.observations: + await catalog_profile_observe.save(session, self.cx.source, result.table, result, job_id=self.job_id, + seq=seq, detected=detected) + await catalog_profile_observe.record_coverage_drops(session, self.cx.source, result.table, result, + detected=detected) async def _finish(self, session: AsyncSession) -> None: """剖析完：万一还有没收尾的表（进度回调写库失败之类）记失败；有 failed、partial、cancelled 的表是 partial，否则 diff --git a/backend/app/data/catalog_profile_observe.py b/backend/app/data/catalog_profile_observe.py index bebeef03..ac5d4968 100644 --- a/backend/app/data/catalog_profile_observe.py +++ b/backend/app/data/catalog_profile_observe.py @@ -224,7 +224,9 @@ async def _codes(cx: Any, table: str, column: str, meta: Mapping[str, Any], size text = catalog.column_family(cp._column_meta(dict(meta), str(day))) != "time" sql = d.window_value_counts_sql(t, c, d.column(dict(meta), str(day)), since, n, text=text) else: - percent = sample_percent(size.rows, max(cx.settings.max_scan_rows, cx.settings.sample_size)) + # 数到上限、只知道「至少多少行」的按下限取档（取到的档偏大，多看一点，不会少看） + percent = sample_percent(size.rows if size.rows is not None else size.at_least, + max(cx.settings.max_scan_rows, cx.settings.sample_size)) sampled = d.sample_value_counts_sql(t, c, percent, n) if percent else None if sampled is not None: basis, sql = {"method": "sample", "percent": percent}, sampled @@ -382,6 +384,45 @@ async def save(session: AsyncSession, source: Any, table: str, result: Any, *, j return fresh +def coverage_drops(findings: Iterable[Mapping[str, Any]]) -> list[dict[str, Any]]: + """关系核对的结论里覆盖率掉到有确证的线以下的（以前够、这次不够）：[{path, target, before, after}]。剖析本身照旧按 + 规则降回推断（或只给人工确认的补上新的覆盖率），这里只把「覆盖率下降」记成一条取值变化。""" + out: list[dict[str, Any]] = [] + line = catalog.PROFILE_VERIFY_COVERAGE + for f in findings: + if not isinstance(f, Mapping) or f.get("kind") != "relation": + continue + before, after = f.get("previous_coverage"), f.get("coverage") + if isinstance(before, (int, float)) and isinstance(after, (int, float)) and before >= line > after: + out.append({"path": f.get("path"), "target": f.get("target"), "before": before, "after": after, + "confirmed": bool(f.get("confirmed"))}) + return out + + +async def record_coverage_drops(session: AsyncSession, source: Any, table: str, result: Any, *, detected: Any) -> int: + """这张表这次剖析里覆盖率掉下来的关系记成 coverage_drop（不提交，和剖析结果同一个事务）。人点的剖析也记：变化记录是 + 全站一份。出错只记日志。""" + from app.data import source_changes as sc + + try: + fresh = 0 + for drop in coverage_drops(getattr(result, "findings", None) or []): + _row, new = await sc.record(session, source, sc.NewChange( + kind=sc.COVERAGE_DROP, table=table, severity="warning", + before={"coverage": drop["before"], "path": drop["path"], "target": drop["target"]}, + after={"coverage": drop["after"], "confirmed": drop["confirmed"]}, + basis={"method": "relation_sample", "key": "relation"}, + affects={"catalog": [{"table": table, "path": drop["path"], "label": drop["target"], + "status": "confirmed" if drop["confirmed"] else "proposed", + "source": None, "needs_action": False}], "templates": [], "rules_touched": False}, + key=(drop["path"], round(float(drop["after"]), 3))), detected) + fresh += int(new) + return fresh + except Exception: # noqa: BLE001 + logger.exception("记录 %s 的关联覆盖率下降时出错", table) + return 0 + + # ========================================================================== # 观察到期、给助手的观察结论 # ========================================================================== @@ -438,5 +479,5 @@ async def observed_facts(session: AsyncSession, source_id: str, __all__ = ["KIND_LABEL", "OBSERVE_CODES_MAX", "OBSERVE_EVERY_H", "OBSERVE_KINDS", "OBSERVER", "SAMPLE_LADDER", - "WINDOW_DAYS", "basis_key", "basis_text", "code_columns", "execute", "last_observed", "observe_reason", - "observed_facts", "plan_observations", "sample_percent", "save"] + "WINDOW_DAYS", "basis_key", "basis_text", "code_columns", "coverage_drops", "execute", "last_observed", "observe_reason", + "observed_facts", "plan_observations", "record_coverage_drops", "sample_percent", "save"] diff --git a/backend/app/data/source_changes.py b/backend/app/data/source_changes.py index cb9de4f5..876bcdf2 100644 --- a/backend/app/data/source_changes.py +++ b/backend/app/data/source_changes.py @@ -374,10 +374,13 @@ def resolve(row: SourceChange, *, action: str, by: str | None, note: str | None return row -def _catalog_paths(row: SourceChange) -> list[dict[str, Any]]: - """影响面里这张表上的目录项（清理目录时删的就是它们）。""" - items = ((row.affects or {}).get("catalog") or []) - return [i for i in items if isinstance(i, Mapping) and i.get("table") == row.table_name and i.get("path")] +def _catalog_paths(row: SourceChange) -> dict[str, list[str]]: + """影响面里的目录项，按表分（清理目录时删的就是它们）：这张表上的，和别的表指向它的关联关系。""" + out: dict[str, list[str]] = {} + for item in (row.affects or {}).get("catalog") or []: + if isinstance(item, Mapping) and item.get("table") and item.get("path"): + out.setdefault(str(item["table"]), []).append(str(item["path"])) + return out def without_paths(notes: Mapping[str, Any], paths: Iterable[str]) -> dict[str, Any]: @@ -397,8 +400,8 @@ async def handle(session: AsyncSession, source: Any, row: SourceChange, action: note: str | None = None, codes: Mapping[str, str] | None = None) -> dict[str, Any]: """执行一个处理动作并记下 resolution（提交）。返回 {change, catalog_version?, job?}。 - - clean_catalog：把影响面里这张表上说到这些列的项从目录里去掉（整份写入，乐观锁），记目录版本。影响面里没有目录项的 - 抛 ChangeInvalid。 + - clean_catalog：把影响面里说到这些列（这张表）的目录项去掉——这张表上的、别的表指向它的关联关系——逐表整份写入 + （乐观锁），记这张表的目录版本（别的表也动了的另记 catalog_versions）。影响面里没有目录项的抛 ChangeInvalid。 - add_codes：码值的新取值加进目录（catalog.save_patch，含义待填写；依据写「数据源变化 #编号」），只给 codes_new。codes 给了就按它加（{码: 含义}，含义可以空），不给就加这条变化测到的新码值。 - redraft：建一个只含这张表的起草任务；reprofile：建一个只含这张表的剖析任务。建不了（模型没配、剖析没开、同源正忙） @@ -418,22 +421,29 @@ async def handle(session: AsyncSession, source: Any, row: SourceChange, action: job_id: str | None = None table = row.table_name if action == "clean_catalog": - paths = [str(i["path"]) for i in _catalog_paths(row)] - if not paths: + by_table = _catalog_paths(row) + if not by_table: raise ChangeInvalid("这条变化的影响面里没有目录项，不需要清理") - entry = await catalog.read_entry(session, source.id, table) - if entry is None: + versions: dict[str, int] = {} + for owner, paths in by_table.items(): + entry = await catalog.read_entry(session, source.id, owner) + if entry is None: + continue + written = await catalog.write_entry(session, source.id, owner, without_paths(entry.notes, paths), + if_version=entry.version, actor=actor, commit=False) + versions[owner] = written.version + if not versions: raise ChangeInvalid(f"{table} 已经没有数据目录了") - notes = without_paths(entry.notes, paths) - written = await catalog.write_entry(session, source.id, table, notes, if_version=entry.version, actor=actor, - commit=False) - version = written.version + # 记这张表的目录版本；别的表（指向它的关联关系）也动了的，另记在 catalog_versions 里 + version = versions.get(table, next(iter(versions.values()))) + if len(versions) > 1: + out["catalog_versions"] = versions elif action == "add_codes": if row.kind != CODES_NEW or not row.column_name: raise ChangeInvalid("只有「出现了新码值」能加入码值") wanted = dict(codes) if codes else {str(c): "" for c in ((row.after or {}).get("codes") or [])} if not wanted or not all(isinstance(k, str) and isinstance(v, str) for k, v in wanted.items()): - raise ChangeInvalid("要加入的码值应为 {码值: 含义} 的对照，含义可以先空着") + raise ChangeInvalid("要加入的码值应为「码值 → 含义」的对照，含义可以先空着") entry = await catalog.read_entry(session, source.id, table) current = entry.version if entry else 0 meta_tables = (getattr(source, "schema_cache", None) or {}).get("tables") or None @@ -453,6 +463,8 @@ async def handle(session: AsyncSession, source: Any, row: SourceChange, action: job_id = await _redraft(session, source, table, actor, row.id) out["job"] = {"kind": "draft", "id": job_id} resolve(row, action=action, by=actor, note=note, catalog_version=version, job_id=job_id) + if "catalog_versions" in out: + row.resolution = {**(row.resolution or {}), "catalog_versions": out["catalog_versions"]} await session.commit() if version is not None: out["catalog_version"] = version diff --git a/backend/tests/test_catalog_observe.py b/backend/tests/test_catalog_observe.py index cd243cdd..73260c8b 100644 --- a/backend/tests/test_catalog_observe.py +++ b/backend/tests/test_catalog_observe.py @@ -289,3 +289,14 @@ async def test_observation_is_not_the_last_profile_and_reaches_the_table_detail( "观察之后才加遮罩的列：渲染那一刻滤掉" detail = schema_view.table_detail(row, "orders", profile=facts) assert f"最新业务日期 paid_on 到 {latest}" in detail + + +def test_coverage_drop_only_when_crossing_the_verified_line(): + """关联覆盖率从够有确证的线（95%）掉到线以下才记 coverage_drop；没有以前的覆盖率、没掉过线的不记。""" + drops = observe.coverage_drops([ + {"kind": "relation", "path": "relations.r1", "target": "a → b", "previous_coverage": 0.99, "coverage": 0.6}, + {"kind": "relation", "path": "relations.r2", "target": "c → d", "previous_coverage": 0.99, "coverage": 0.97}, + {"kind": "relation", "path": "relations.r3", "target": "e → f", "previous_coverage": None, "coverage": 0.5}, + {"kind": "codes", "path": "columns.x.codes"}, + ]) + assert [(d["path"], d["before"], d["after"]) for d in drops] == [("relations.r1", 0.99, 0.6)] diff --git a/backend/tests/test_source_maintenance.py b/backend/tests/test_source_maintenance.py index 1f7af490..468c580c 100644 --- a/backend/tests/test_source_maintenance.py +++ b/backend/tests/test_source_maintenance.py @@ -208,3 +208,28 @@ async def test_steps_api_and_check_target(client, tmp_path): assert await sm.check_target(session, sid) == sm.NOT_MAINTAINED finally: await drop_source(sid) + + +async def test_scheduler_switched_off_never_fires_maintenance(client, tmp_path, made): + """AGENTLAB_SCHEDULER 关着（测试环境就是关着的，见 conftest）：调度循环不起，到点的维护不会触发——拷来的数据目录不替原库 + 查业务库。照实说：这时没有触发记录（定时只有一套，没起循环就没有「这一拍」）。""" + import asyncio + + from app import scheduler + + sid = await _source(client, tmp_path / "f.db") + try: + schedule_id = await _schedule(sid, next_run_at=utcnow() - timedelta(minutes=1)) + made.append(schedule_id) + assert scheduler.scheduler_enabled() is False + await scheduler.start() + await asyncio.sleep(0.3) + await scheduler.stop() + async with SessionLocal() as session: + fires = list((await session.execute(select(ScheduleFire).where( + ScheduleFire.schedule_id == schedule_id))).scalars()) + jobs = list((await session.execute(select(IntrospectJob).where( + IntrospectJob.source_id == sid, IntrospectJob.trigger == "schedule"))).scalars()) + assert fires == [] and jobs == [] + finally: + await drop_source(sid) diff --git a/scripts/dialect-it.sh b/scripts/dialect-it.sh index 0b7319ec..26992331 100755 --- a/scripts/dialect-it.sh +++ b/scripts/dialect-it.sh @@ -1,8 +1,8 @@ #!/usr/bin/env bash # 方言的容器集成测试（下一轮 03 细案 C2）：本机 docker 起 PostgreSQL 16、MySQL 8.4、MariaDB 11、Oracle Free 23 四个容器， # 建虚构的夹具（保留字列、Oracle 小写带引号的表、CHAR 码值、MySQL ENUM、零日期 / 无穷大 / 占位日期、做旧的统计信息、 -# 超过整表统计上限的大表、带索引的日期列），用 AgentLab 自己的探查和剖析代码跑全部剖析查询和取值观察的新方法， -# 断言零报错、结论和手算的一致。 +# 超过整表统计上限的大表、带索引的日期列），用 AgentLab 自己的探查和剖析代码跑全部剖析查询、对象索引的字典查询 +# （第 2 波）和取值观察的四类检查（第 3 波），断言零报错、结论和手算的一致。 # # 为什么不进 CI：要拉几 GB 的镜像、Oracle 起一次要一两分钟。要用户同意才在本机跑。跑完删容器、留镜像。 # diff --git a/scripts/dialect_it.py b/scripts/dialect_it.py index ff153dd1..86538338 100644 --- a/scripts/dialect_it.py +++ b/scripts/dialect_it.py @@ -5,8 +5,12 @@ 2. 用 AgentLab 自己的探查（app.data.introspect）读结构，核对类型写法、ENUM 取值、Oracle 小写带引号的名字、索引首列； 3. 用 AgentLab 自己的剖析（catalog_profile.profile_table，目录读写换成内存里的）逐表剖析，经只读连接、守卫、单条时限， 和线上走的是同一条路； -4. 直接跑取值观察要用的新查询（按日行数、时间窗口里的取值分布、块采样）； -5. 断言：所有查询零报错（留痕里没有超时、失败、守卫拒），每个结论和下面按夹具定义手算的一致。 +4. 直接跑取值观察要用的新查询（按日行数、时间窗口里的取值分布、块采样、最新业务日期）； +5. 对象索引的字典查询（第 2 波，data/source_objects.list_objects：MySQL information_schema.TABLES、PostgreSQL pg_class 加 + obj_description、Oracle ALL_OBJECTS 加 ALL_TAB_COMMENTS）：走字典查询（不退回 Inspector）、类型、注释、行数估算对得上； +6. 取值观察的四类检查（第 3 波，data/catalog_profile_observe）经剖析引擎跑一遍（只观察，目录读写换成内存里的）：行数、 + 码值观察（小表整表、大表 PG / Oracle 块采样、MySQL 按存储顺序的前 N 行）、最新业务日期（大表业务日期列打头有索引）； +7. 断言：所有查询零报错（留痕里没有超时、失败、守卫拒），每个结论和下面按夹具定义手算的一致。 夹具的数（手算的依据）：orders 40 行，id 1–40。 - store_code = id % 5 + 1（整数子列，指向文字主键 stores.code，'1'–'5' 都在）； @@ -36,7 +40,7 @@ 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 catalog, catalog_profile, introspect, source_objects # noqa: E402 from app.data import engine as data_engine # noqa: E402 from app.data.catalog_profile import ProfileSettings, SqlDialect # noqa: E402 @@ -122,9 +126,13 @@ def _ddl(t: Target) -> list[str]: 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)})", + orders = f"CREATE TABLE orders ({', '.join(cols)})" + if t.mysql: + orders += " COMMENT='订单'" + out = [f"CREATE TABLE stores (code {vc}(10) PRIMARY KEY, name {vc}(20))", orders, "CREATE INDEX ix_orders_paid_at ON orders (paid_at)"] + if not t.mysql: + out.append("COMMENT ON TABLE orders IS '订单'") events = f"CREATE TABLE events (id {intg} PRIMARY KEY, kind {vc}(10))" if t.pg: events += " WITH (autovacuum_enabled = false)" @@ -218,7 +226,8 @@ async def _setup(t: Target) -> None: # ---------------------------------------------------------------- 剖析：目录换成内存里的 -async def _profile(source: Any, tables: list[str], notes: dict[str, dict[str, Any]]) -> dict[str, Any]: +async def _profile(source: Any, tables: list[str], notes: dict[str, dict[str, Any]], *, + observe: bool = False) -> dict[str, Any]: cx = catalog_profile.make_context(source, SETTINGS, options={}) captured: dict[str, Any] = {} @@ -233,7 +242,7 @@ async def _profile(source: Any, tables: list[str], notes: dict[str, dict[str, An raise AssertionError(f"剖析要了没预料的库操作：{fn}") for table in tables: - await catalog_profile.profile_table(cx, table, db=db) + await catalog_profile.profile_table(cx, table, db=db, regular=not observe, observe=observe) return captured @@ -359,6 +368,50 @@ async def run_target(t: Target) -> None: check(total <= N_EVENTS_OLD + N_EVENTS_NEW, f"块采样跑得通（取到 {total} 行）") else: check(sampled is None, "MySQL / MariaDB 没有块采样，照实返回 None") + # ---- 对象索引的字典查询（第 2 波） + index = await source_objects.list_objects(source, None) + found = {o.name: o for o in index.objects} + check(index.method == "dictionary" and index.note is None, + f"对象索引走字典查询，没有退回 Inspector：{index.method} {index.note or ''}") + wanted = {"orders", "events", "stores"} | ({"visits"} if t.oracle else set()) + check(wanted <= set(found), f"对象索引列出了夹具表：{sorted(found)}") + check(found["orders"].kind == "table" and found["orders"].comment == "订单", + f"对象索引的类型、表注释：{found.get('orders')}") + if t.oracle: + check(found["visits"].quote is True, "Oracle 小写带引号的表在对象索引里记了 quote") + est = found["events"].est_rows + check(est is None or est > 0, f"对象索引的行数估算（做过统计的表）：{est}") + + # ---- 取值观察（第 3 波）：只观察，经剖析引擎、只读守卫、单条时限 + paid = _key(orders, "paid_at") + obs_notes = { + "orders": {"business_date": {"value": {"column": paid}, "source": "human", "status": "confirmed"}, + "columns": {_key(orders, "status"): {"codes": {"value": {"A": "", "B": ""}, "source": "human", + "status": "confirmed", "complete": True}}}}, + "events": {"columns": {_key(events, "kind"): {"codes": {"value": {"open": "", "close": ""}, + "source": "profile", "status": "proposed"}}}}, + } + seen = await _profile(source, ["orders", "events"], obs_notes, observe=True) + logs = [q for _, r in seen.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 "")) + got_o = {(o["kind"], o["column"].lower()): o for o in seen["orders"][1].observations} + got_e = {(o["kind"], o["column"].lower()): o for o in seen["events"][1].observations} + check(got_o.get(("rows", ""), {}).get("data", {}).get("rows") == N_ORDERS, f"orders 的行数观察：{got_o.get(('rows', ''))}") + codes = got_o.get(("codes_observe", "status")) + check(codes is not None and codes["basis_key"] == "full" + and Counter(dict(map(tuple, codes["data"]["values"]))) == EXPECT["status"], + f"小表的码值观察看整表、和手算一致：{codes and codes['data']}") + fresh = got_o.get(("freshness", "paid_at")) + check(fresh is not None and fresh["data"]["max"][:19].replace("T", " ") == str(EXPECT["paid_at"][1]), + f"最新业务日期排除了无效日期、和手算一致：{fresh and fresh['data']}") + ev = got_e.get(("codes_observe", "kind")) + method = "sample" if (t.pg or t.oracle) else "first_rows" + check(ev is not None and ev["basis"]["method"] == method, + f"大表、没有业务日期：{'块采样' if method == 'sample' else '按存储顺序的前 N 行'}（{ev and ev['basis_key']}）") + res = await data_engine.run_query(source, d.max_date_sql(tbl, paid_col)) + check(bool(res.rows) and str(res.rows[0][0])[:19].replace("T", " ") == str(EXPECT["paid_at"][1]), + f"max_date_sql 直接跑：{res.rows}") finally: await data_engine.engines.invalidate(source.id)