From c95e5b4e84178dbee3871a9504b4f479d7d4fc1d Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Fri, 16 Oct 2026 16:40:41 +0800 Subject: [PATCH] =?UTF-8?q?W3-a=20=E5=8F=96=E5=80=BC=E8=A7=82=E5=AF=9F?= =?UTF-8?q?=EF=BC=9A=E5=89=96=E6=9E=90=E5=BC=95=E6=93=8E=E7=9A=84=E5=9B=9B?= =?UTF-8?q?=E7=B1=BB=E8=A7=82=E5=AF=9F=E6=A3=80=E6=9F=A5=E3=80=81=E6=AF=94?= =?UTF-8?q?=E8=BE=83=E5=99=A8=E3=80=81=E6=92=A4=E9=94=80=E3=80=8C=E5=B7=B2?= =?UTF-8?q?=E5=88=97=E5=85=A8=E3=80=8D=EF=BC=88next-round=2003=20B2?= =?UTF-8?q?=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 码值出现新取值、比例剧变、业务日期断档、某天没数据，以前都看不到：确认过的码值列不再取分布，大表只看物理顺序最老的 前 10 万行，「已列全」过期以后 SQL 检查还照它提示。现在： - 剖析引擎加四类观察检查（data/catalog_profile_observe.py）：rows（行数，用剖析开头那条估行数的结果）、codes_observe （所有码值列，包括人工确认过、标了已列全的）、freshness（已确认或已验证的业务日期的最新日期）、daily_volume（最近 35 天按天计数）。计入每张表的预算、止损和留痕，每条观察记口径 basis 和它出自哪几条查询；遮罩列不进任何观察查询。 - 大表口径照目录-D7：有确认的业务日期且该列打头有索引看 35 天窗口；PG / Oracle 没有就块采样（比例按表的大小取一档， 不跨档比）；MySQL / SQLite 没有就按存储顺序的前 N 行，照实写口径。最新业务日期、按日行数在大表上只在有索引时才查。 SqlDialect 加 max_date_sql（有效区间里的最大日期），方言金样跟着加了两条。 - data/value_drift.py 纯函数比较器：新码值（已列全的任何新值、没列全的要占 0.5% 以上且有历史）、PSI 剧变（≥0.25 warning、0.1–0.25 info，两次都要 1,000 行）、断档（超过 max(2×中位间隔, 3 天)）、缺天（以前的窗口没有过）、量跌 （不到前 28 天中位数的 30%）、行数跌 30% / 涨 3 倍（估算值要连续两次）。只和同口径的历史比。 - 观察记进 table_observations（每张表每种观察留最近 26 条），比出的取值变化进 source_changes，和这张表的剖析结果、目录在 同一个事务里。机器唯一写进目录的事：数据证明码值表不再列全时撤销「已列全」（目录版本 +1，署名「取值观察」，值、含义、 状态都不动），SQL 检查随之不再按旧码值表提示 unknown_code。 - 剖析任务的范围加 observe（观察到期的有目录的表，只观察）和 maintenance（定时维护：需要剖析的 + 观察到期的）；人点的 「需要剖析 / 全部 / 指定的表」照旧不观察。只观察的那一项不算「上次剖析」（不顶掉表清单的上次剖析和给助手的剖析事实）。 任务的设置快照里记 trigger / fire_at / schedule_id（剖析任务表没有这几列）。 - 观察结论（最新业务日期、更新的行数）经 last_facts → schema_view.profile_lines 和剖析事实同一节进 look、db_schema、 /schema、表助手（X7），遮罩在渲染那一刻再核。 测试：test_value_drift（阈值两侧、口径不同不比）、test_catalog_observe（四类观察与口径、撤销已列全与 SQL 检查、去重、 遮罩不进查询、大表窗口与前 N 行、PG / Oracle 采样 SQL、观察不算上次剖析、进表详情）。 --- backend/app/data/catalog_profile.py | 57 ++- backend/app/data/catalog_profile_jobs.py | 158 ++++++- backend/app/data/catalog_profile_observe.py | 442 ++++++++++++++++++ backend/app/data/schema_view.py | 21 +- backend/app/data/value_drift.py | 263 +++++++++++ backend/tests/fixtures/dialect/mariadb.sql | 4 + backend/tests/fixtures/dialect/mysql.sql | 4 + backend/tests/fixtures/dialect/oracle.sql | 4 + backend/tests/fixtures/dialect/postgresql.sql | 4 + backend/tests/fixtures/dialect/sqlite.sql | 4 + backend/tests/test_catalog_observe.py | 291 ++++++++++++ backend/tests/test_profile_dialect_golden.py | 5 +- backend/tests/test_value_drift.py | 114 +++++ 13 files changed, 1348 insertions(+), 23 deletions(-) create mode 100644 backend/app/data/catalog_profile_observe.py create mode 100644 backend/app/data/value_drift.py create mode 100644 backend/tests/test_catalog_observe.py create mode 100644 backend/tests/test_value_drift.py diff --git a/backend/app/data/catalog_profile.py b/backend/app/data/catalog_profile.py index f4976de7..7198e95d 100644 --- a/backend/app/data/catalog_profile.py +++ b/backend/app/data/catalog_profile.py @@ -28,6 +28,10 @@ hooks 回调（后台任务据此把进度落库），整张表结束把结论 - 数据源设置里遮罩的列不取样，也不出现在任何一条剖析查询里； - 只取低基数列的取值分布（最多 21 个取值），不取明细行。 +**取值观察**（next-round 03 B2，data/catalog_profile_observe）是这里的另外四类检查（行数、码值观察、最新业务日期、按日 +行数），同样计入每张表的预算、止损和留痕，遮罩列同样不出现在任何一条查询里；只在剖析任务的范围是 observe / maintenance +时带上（profile_table 的 observe）。 + 表助手的只读查询、起草时核对有效记录条件走单条查询的入口（run_single_query）：同一个只读守卫、同一个单条时限，一次一条， 不设预算、不留痕。 """ @@ -480,7 +484,14 @@ class SqlDialect: f"SUM(CASE WHEN {c} < {lo} OR {c} >= {hi} THEN 1 ELSE 0 END)") return f"SELECT {', '.join(parts)} FROM {table}" - # ---- 取值观察（下一轮第 3 波用）：窗口、按日行数、块采样。这里先把各方言的写法定下来，金样和容器集成测试都覆盖 + # ---- 取值观察（data/catalog_profile_observe）：最新业务日期、窗口、按日行数、块采样。金样和容器集成测试都覆盖 + + def max_date_sql(self, table: str, column: str, *, text: bool = False) -> str: + """业务日期列的最新日期：有效区间（VALID_DATE_FROM ≤ x < VALID_DATE_BEFORE）里的最大值，占位日期、无穷大不算。大表 + 只在这一列打头有索引时发（取最大值走索引）。""" + lo = self.date_literal(VALID_DATE_FROM, text=text) + hi = self.date_literal(VALID_DATE_BEFORE, text=text) + return f"SELECT MAX({column}) FROM {table} WHERE {column} >= {lo} AND {column} < {hi}" def _window(self, column: str, since: Any, text: bool) -> str: """时间窗口的条件：since 起（含），并且早于占位日期 9999-12-31——PostgreSQL 的 infinity、Oracle 的 9999-12-31 @@ -829,7 +840,8 @@ class ProfileDisabled(Exception): class ProfileSkip: """没做的一项和原因。 - kind：relation 关系、codes 码值、date 日期列、row_estimate 行数估算、table 整张表。 + kind：relation 关系、codes 码值、date 日期列、row_estimate 行数估算、table 整张表；取值观察的 rows 行数、codes_observe + 码值观察、freshness 最新业务日期、daily_volume 按日行数（data/catalog_profile_observe）。 reason：budget 这张表的查询次数用完、deadline 这张表的时长用完、failed 连续失败止损、slow 超时比例止损、 cancelled 停止或服务重启、timeout 超时、error 查询失败、rejected 未通过守卫（以上算「运行原因没查成」，见 UNFINISHED_REASONS）；masked 字段被遮罩、too_large 表太大、view 视图、unsupported 暂不支持、no_data 没有数据、 @@ -875,6 +887,8 @@ class TableResult: version: int = 0 #: 整张表的原因：表结构里没有、视图、空表、没有可剖析的项、写入一直冲突、停止 / 中断 / 止损 error: str | None = None + #: 取值观察（data/catalog_profile_observe）观察到的：[{kind, column, basis, basis_key, data}]，由剖析任务和结果一起落库 + observations: list[dict[str, Any]] = field(default_factory=list) @property def skipped_count(self) -> int: @@ -1018,6 +1032,8 @@ class _TableCx: self.budget = TableBudget(job.settings) #: 正在做的检查项（result.checks 里的那一个）；None 是这张表开头的估行数 self.check: dict[str, Any] | None = None + #: 这张表此刻的目录（开始剖析这张表时读的）：取值观察按它找已确认的业务日期 + self.table_notes: dict[str, Any] = {} def is_masked(self, column: str) -> bool: return column.lower() in self.job.masked @@ -1767,8 +1783,16 @@ def pending_relation_paths(notes: dict[str, Any] | None) -> set[str]: ESTIMATING = "正在估算 {table} 的行数" +#: 取值观察各项进行中的说法 +OBSERVE_CURRENT = {"rows": "正在记 {table} 的行数", "codes_observe": "正在观察 {table}.{column} 的取值分布", + "freshness": "正在取 {table}.{column} 的最新业务日期", + "daily_volume": "正在数 {table} 最近 35 天每天的行数"} + + def _current_text(table: str, check: PlannedCheck) -> str: """进行中时正在做哪一项的整句（逐表一行的最后一栏）。""" + if check.kind in OBSERVE_CURRENT: + return OBSERVE_CURRENT[check.kind].format(table=table, column=check.column or "") if check.kind == "relation": return f"正在核对 {table}.{check.target}" if check.kind == "codes": @@ -1805,7 +1829,11 @@ def _stop_detail(cx: _TableCx, reason: str) -> str: async def _execute(cx: _TableCx, table: str, check: PlannedCheck, out: _TableOutcome) -> None: - if check.kind == "relation": + from app.data import catalog_profile_observe as observe + + if check.kind in observe.OBSERVE_KINDS: + await observe.execute(cx, table, check, cx.result) + elif check.kind == "relation": await _check_relation(cx, table, check.relation or {}, out, cx.result) elif check.kind == "codes": await _check_codes(cx, table, str(check.column), out, cx.result) @@ -1834,6 +1862,14 @@ async def _run_checks(cx: _TableCx, table: str, plan: list[PlannedCheck], runnab finally: result.current = None if size.rows == 0: + # 空表：行数观察照记 0 行（一张表变空是最大的「行数下跌」），别的项都不查 + for check, planned in zip(result.checks, plan): + if planned.kind == "rows" and check["state"] == "pending": + result.observations.append({"kind": "rows", "column": "", "basis": {"method": size.method, + "query_refs": []}, + "basis_key": str(size.method), "data": {"rows": 0, "at_least": None, + "method": size.method}}) + check["state"] = "done" _close_open(result, plan, out, "no_data", f"{table} 是空表，无法剖析") result.status, result.error = "skipped", EMPTY_ERROR return @@ -1875,9 +1911,13 @@ async def _read_notes(session: AsyncSession, source_id: str, table: str) -> dict async def profile_table(cx: ProfileContext, table: str, *, db: Callable[..., Awaitable[Any]] | None = None, hooks: TableHooks | None = None, - actor: str | None = None) -> TableResult: + actor: str | None = None, regular: bool = True, observe: bool = False) -> TableResult: """剖析一张表并把结论写进目录，返回这张表的结果（TableResult）。 + regular：做关系、码值、日期这几项剖析（缺省）；observe：接着做取值观察的四类检查（data/catalog_profile_observe，观察到的 + 记在 result.observations，由剖析任务和结果一起落库）。只观察（regular=False）时不往目录里并剖析结论——不然这次没做的剖析 + 项会被当成「这次完整检查过、却没再得出」删掉；目录里唯一可能变的是观察撤销「已列全」。 + - 表结构里没有、视图：skipped，不发查询。 - 先列全检查项（plan_checks，读的是这时目录里的最新内容），按规则不查的直接记跳过；一项可查的都没有：skipped 「没有可剖析的项」，不发查询。 @@ -1905,9 +1945,14 @@ async def profile_table(cx: ProfileContext, table: str, *, await db(_record, cx.source.id, table, None, result, actor, hooks.recorded) return result notes = await db(_read_notes, cx.source.id, table) - plan = plan_checks(table, meta, notes, tables=cx.tables, masked=cx.masked, outside=cx.outside) + plan = plan_checks(table, meta, notes, tables=cx.tables, masked=cx.masked, outside=cx.outside) if regular else [] + if observe: + from app.data import catalog_profile_observe + + plan += catalog_profile_observe.plan_observations(table, meta, notes, masked=cx.masked) out = _TableOutcome() tcx = _TableCx(cx, table, result, hooks) + tcx.table_notes = notes result.checks = [p.as_check() for p in plan] for planned in plan: if planned.skip is not None: @@ -1939,7 +1984,7 @@ async def profile_table(cx: ProfileContext, table: str, *, result.current = None result.row_estimate = cx.sizes.get(table) result.elapsed_ms = int((time.monotonic() - started) * 1000) - await db(_record, cx.source.id, table, out, result, actor, hooks.recorded) + await db(_record, cx.source.id, table, out if regular else None, result, actor, hooks.recorded) if stopped is not None: raise stopped return result diff --git a/backend/app/data/catalog_profile_jobs.py b/backend/app/data/catalog_profile_jobs.py index f7160c3a..4f31887d 100644 --- a/backend/app/data/catalog_profile_jobs.py +++ b/backend/app/data/catalog_profile_jobs.py @@ -27,6 +27,12 @@ data/catalog_job_runtime.py），逐表剖析（引擎在 data/catalog_profile.p 启动时还一次性迁移剖析设置的旧键（catalog_profile.migrate_settings）。 - **保留**（prune）：每个源保留最近 KEEP_JOBS 个结束了的剖析任务，更早的删掉；某张表「上次剖析」所在的任务不删，表详情的 「上次剖析」一直在。 +- **取值观察**（next-round 03 B2，data/catalog_profile_observe）：范围另有两种——observe（有数据目录、观察到期的表，只做 + 观察检查）和 maintenance（定时维护用：需要剖析的表做剖析，观察到期的表做观察，两样都要的两样都做）。每张表做哪几样按建 + 任务时列出的检查项定；观察到的随这张表的结果落库（同一个事务），和同口径的历史比出的取值变化写进 source_changes。人点的 + 「需要剖析 / 全部 / 指定的表」照旧不观察。 +- **触发**：任务的设置快照里另记 trigger（manual 人点的 / schedule 定时维护）、fire_at、schedule_id（剖析任务表没有这几列、 + 这一轮不加列，见 trigger_of），接口按 trigger、fire_at 给。 - **读**：job_dict（任务和逐表一行）、task_detail（一张表的检查项、发现、跳过的项、发过的查询）、last_profiles / last_detail （表清单每行、表详情的「上次剖析」）、due_reason（这张表要不要剖析、为什么）。「上次剖析」是真正开始过的最近一次（见 _ranked）：没开始就停了的、还排着的那一项不挡掉上一次真正剖析的行数、发过的查询。 @@ -200,6 +206,23 @@ def _counts(statuses: list[str]) -> dict[str, int]: return counts +#: 剖析任务的范围：selected 指定的表、needed 需要剖析的表、all 全部表；observe 观察到期的表（只做取值观察）、maintenance +#: 定时维护（需要剖析的表 + 观察到期的表） +SCOPES = ("selected", "needed", "all", "observe", "maintenance") +#: 带取值观察的范围 +OBSERVING_SCOPES = ("observe", "maintenance") +#: 设置快照里记触发的几个键（不是剖析设置：ProfileSettings.from_dict 不认它们） +_TRIGGER_KEYS = ("trigger", "fire_at", "schedule_id") + + +def trigger_of(job: Any) -> dict[str, Any]: + """这个任务是怎么建的：{trigger: manual|schedule, fire_at, schedule_id}。剖析任务表没有这几列（只加不改的迁移这一轮 + 不加），记在设置快照里。""" + raw = getattr(job, "settings", None) or {} + return {"trigger": raw.get("trigger") or "manual", "fire_at": raw.get("fire_at"), + "schedule_id": raw.get("schedule_id")} + + def job_dict(job: CatalogProfileJob, tasks: list[Any], *, waiting_for: str | None = None) -> dict[str, Any]: """接口返回的任务（CatalogProfileJob）：逐表按剖析顺序；计数、合计都按逐表的算。""" rows = [task_row(t) for t in tasks] @@ -219,6 +242,8 @@ def job_dict(job: CatalogProfileJob, tasks: list[Any], *, waiting_for: str | Non "finished_at": _iso(job.finished_at), "created_by": job.created_by, "retry_of": job.retry_of, + "trigger": trigger_of(job)["trigger"], + "fire_at": trigger_of(job)["fire_at"], "waiting_for": waiting_for, "total": len(rows), "done": sum(counts.get(s, 0) for s in TASK_DONE), @@ -309,6 +334,26 @@ _NEWEST = (CatalogProfileJob.created_at.desc(), CatalogProfileJob.id.desc(), Cat _SHOWN = (case((CatalogProfileTask.started_at.is_(None), 1), else_=0), *_NEWEST) +#: 只做了取值观察的那一项（observe 范围、maintenance 范围里只到期观察的表）在逐表一行的 summary 里带这个标记：它不是一次 +#: 剖析，不算「上次剖析」——不然表清单的「上次剖析」、给助手的剖析事实（行数、日期范围）会被一次只记了行数的观察顶掉 +OBSERVE_ONLY = "observe_only" +_PROFILED = func.coalesce(func.json_extract(CatalogProfileTask.summary, f"$.{OBSERVE_ONLY}"), 0) == 0 + + +def _observe_only(checks: list[Any] | None) -> bool: + from app.data import catalog_profile_observe as observe + + kinds = {c.get("kind") if isinstance(c, dict) else getattr(c, "kind", None) for c in checks or []} + return bool(kinds) and kinds <= set(observe.OBSERVE_KINDS) + + +def _summary_of(result: TableResult) -> dict[str, Any]: + out: dict[str, Any] = dict(result.summary()) + if _observe_only(result.checks): + out[OBSERVE_ONLY] = 1 + return out + + def _ranked(source_id: str, *columns: Any) -> Any: """每张表剖析过的各项：子查询，带两个排名。 @@ -321,7 +366,7 @@ def _ranked(source_id: str, *columns: Any) -> Any: shown = func.row_number().over(partition_by=CatalogProfileTask.table_name, order_by=_SHOWN).label("shown") latest = func.row_number().over(partition_by=CatalogProfileTask.table_name, order_by=_NEWEST).label("latest") return (select(*columns, shown, latest).join(CatalogProfileJob, CatalogProfileJob.id == CatalogProfileTask.job_id) - .where(CatalogProfileJob.source_id == source_id)).subquery() + .where(CatalogProfileJob.source_id == source_id, _PROFILED)).subquery() async def last_profiles(session: AsyncSession, source_id: str) -> dict[str, dict[str, Any]]: @@ -365,7 +410,8 @@ def public_last(last: dict[str, Any] | None) -> dict[str, Any] | None: async def last_facts(session: AsyncSession, source_id: str, tables: list[str] | None = None) -> dict[str, dict[str, Any]]: """每张表「上次剖析」（见 _ranked 的 shown，和表清单、表详情同一项）里给助手看的剖析事实：{表名: {at, job_id, - row_estimate, date_ranges, snapshot_id}}，一条查询批量取。没有行数估算也没有日期范围的不给；tables 给了只取这几张。 + row_estimate, date_ranges, snapshot_id, observed?}}，一条查询批量取。没有行数估算也没有日期范围的不给；tables 给了只取 + 这几张。observed 是取值观察最近一次的结论（catalog_profile_observe.observed_facts：最新业务日期、行数），有就给。 look、运行时 db_schema、/schema、表助手四处都经 schema_view.profile_lines 渲染这一份（next-round 遗留 e）：剖析数字 是会过期的观测，不写进数据目录；遮罩、快照在渲染那一刻再核。""" @@ -386,6 +432,12 @@ async def last_facts(session: AsyncSession, source_id: str, at = row["finished_at"] or row["started_at"] out[row["table_name"]] = {"at": _iso(at), "job_id": row["job_id"], "row_estimate": estimate, "date_ranges": ranges, "snapshot_id": row["snapshot_id"]} + # 取值观察的结论（最新业务日期、行数，next-round 03 B2）和剖析事实同一节：只有观察、没剖析过的表也给 + from app.data import catalog_profile_observe + + for table, observed in (await catalog_profile_observe.observed_facts(session, source_id, tables)).items(): + out.setdefault(table, {"at": None, "job_id": None, "row_estimate": None, "date_ranges": [], + "snapshot_id": None})["observed"] = observed return out @@ -394,7 +446,7 @@ async def last_detail(session: AsyncSession, source_id: str, table: str) -> dict found = (await session.execute( select(CatalogProfileTask, CatalogProfileJob) .join(CatalogProfileJob, CatalogProfileJob.id == CatalogProfileTask.job_id) - .where(CatalogProfileJob.source_id == source_id, CatalogProfileTask.table_name == table) + .where(CatalogProfileJob.source_id == source_id, CatalogProfileTask.table_name == table, _PROFILED) .order_by(*_SHOWN).limit(1) )).first() return None if found is None else task_detail(found[1], found[0]) @@ -476,6 +528,8 @@ NOTHING_NEEDED = ("没有需要剖析的表：能剖析的表都剖析过了， "日期列（上次是空表的不算）。要重新剖析，请选「全部表」或指定要剖析的表") NOTHING_AT_ALL = ("没有可以剖析的表：表结构里的表都是视图，或者没有可剖析的项（要核对的关联关系、像码值的列、日期列都没有，" "或者都被遮罩了）。请先起草数据目录，或指定要剖析的表") +NOTHING_TO_OBSERVE = ("没有要观察的表：取值观察只看有数据目录的表，这些表最近 {h} 小时内都观察过了（或者还没有数据目录）") +NOTHING_TO_MAINTAIN = ("没有需要剖析或观察的表：能剖析的表都剖析过了，有数据目录的表最近 {h} 小时内也都观察过了") async def outside_of(session: AsyncSession, row: Any) -> frozenset[str]: @@ -536,6 +590,8 @@ async def _pick(session: AsyncSession, row: Any, source: Any, *, tables: list[st else: picked.append((key, None, None, checks)) return picked + if scope in OBSERVING_SCOPES: + return await _pick_observing(session, source, scope, all_tables, entries, notes, plan, masked) usage = await catalog.table_usage(session, source) last = await last_profiles(session, source.id) if scope == "needed" else {} order = {name: i for i, name in enumerate(all_tables)} @@ -557,6 +613,54 @@ async def _pick(session: AsyncSession, row: Any, source: Any, *, tables: list[st return [(name, None, None, checks) for name, checks in candidates] +async def _pick_observing(session: AsyncSession, source: Any, scope: str, all_tables: dict[str, Any], + entries: dict[str, Any], notes: Any, plan: Any, + masked: set[str]) -> list[tuple[str, str | None, str | None, list[Any]]]: + """observe / maintenance 的表：观察到期的、有数据目录的表做观察检查（catalog_profile_observe.plan_observations）； + maintenance 另加需要剖析的表做剖析（同 needed）。两样都要的两样都做。剖析的在前（有待核实关系的再往前），再按使用次数、 + 表结构顺序。挑出 0 张抛 ProfileNothing。""" + from app.data import catalog_profile_observe as observe + + observed = await observe.last_observed(session, source.id) + last = await last_profiles(session, source.id) if scope == "maintenance" else {} + usage = await catalog.table_usage(session, source) + order = {name: i for i, name in enumerate(all_tables)} + candidates: list[tuple[str, list[Any], bool]] = [] + for name, meta in all_tables.items(): + if not isinstance(meta, dict) or meta.get("is_view"): + continue + regular: list[Any] = [] + if scope == "maintenance": + planned = plan(name) + if any(c.skip is None for c in planned) and due_reason(meta, notes(name), last.get(name), planned): + regular = planned + watched: list[Any] = [] + if name in entries and observe.observe_reason(observed.get(name)) is not None: + watched = observe.plan_observations(name, meta, notes(name), masked=masked) + if regular or watched: + candidates.append((name, regular + watched, bool(regular))) + if not candidates: + text = NOTHING_TO_MAINTAIN if scope == "maintenance" else NOTHING_TO_OBSERVE + raise ProfileNothing(text.format(h=observe.OBSERVE_EVERY_H)) + candidates.sort(key=lambda c: (not c[2], not catalog_profile.has_pending_relations(notes(c[0])), + -usage.get(c[0], 0), order[c[0]])) + return [(name, None, None, checks) for name, checks, _ in candidates] + + +def task_modes(scope: str, checks: list[Any] | None) -> tuple[bool, bool]: + """一张表这次做哪几样：(剖析, 观察)。按建任务（或开始时重挑）列出的检查项：observe 范围只观察；maintenance 看检查项里 + 有没有剖析的、观察的；别的范围只剖析。""" + from app.data import catalog_profile_observe as observe + + kinds = {c.get("kind") for c in checks or [] if isinstance(c, dict)} + watched = bool(kinds & set(observe.OBSERVE_KINDS)) + if scope == "observe": + return False, True + if scope == "maintenance": + return bool(kinds - set(observe.OBSERVE_KINDS)), watched + return True, False + + def _tasks_for(job_id: str, picked: list[tuple[str, str | None, str | None, list[Any]]], now: Any) -> list[CatalogProfileTask]: """_pick 挑出的表 → 逐表的行：按规则不剖析的（表结构里没有、视图、没有可剖析的项）直接记跳过，其余排队。""" @@ -567,8 +671,9 @@ def _tasks_for(job_id: str, picked: list[tuple[str, str | None, str | None, list rule = [ProfileSkip(c.kind, c.target, c.skip[0], c.skip[1], c.path) for c in checks if c.skip] tasks.append(_skipped_task(job_id, seq, name, reason, str(why), shown, rule, now)) continue + summary = {**_empty_summary(), **({OBSERVE_ONLY: 1} if _observe_only(checks) else {})} tasks.append(CatalogProfileTask(job_id=job_id, seq=seq, table_name=name, status="pending", checks=shown, - summary=_empty_summary())) + summary=summary)) return tasks @@ -580,22 +685,31 @@ async def _behind_draft(session: AsyncSession, source_id: str) -> bool: async def create_job(session: AsyncSession, row: Any, source: Any, *, tables: list[str] | None, scope: str, - settings: ProfileSettings, actor: str | None, - retry_of: str | None = None) -> tuple[CatalogProfileJob, list[CatalogProfileTask]]: + settings: ProfileSettings, actor: str | None, retry_of: str | None = None, + trigger: str = "manual", fire_at: str | None = None, schedule_id: str | None = None, + allow_empty: bool = False) -> tuple[CatalogProfileJob, list[CatalogProfileTask]]: """建一个排队的剖析任务（还没开始，调用方接着 manager.start）。row 是数据源记录（遮罩按它的设置），source 是读表结构 用的源（上传源是绑定当前快照的视图）。同源已有排队或进行中的剖析任务抛 ProfileBusy；挑出 0 张表抛 ProfileNothing。 同源有起草任务时照建，排在它后面——这时挑出 0 张表也照建。不管排没排队，这里挑出的表单都是暂定的：开始剖析时按那时的 - 目录再定一遍（_ProfileRun._rescope）。建完顺手清掉超出保留数的旧任务。""" + 目录再定一遍（_ProfileRun._rescope）。建完顺手清掉超出保留数的旧任务。 + + trigger / fire_at / schedule_id 记进设置快照（见 trigger_of）。allow_empty：挑出 0 张也照建（定时维护排在同一次触发的 + 探查后面：探查完了表结构、目录可能就不一样了，开始时再挑）。""" + if scope not in SCOPES: + raise ValueError(f"不认识的剖析范围：{scope}") await ensure_free(session, row, tables) try: picked = await _pick(session, row, source, tables=tables, scope=scope) except ProfileNothing: # 排在起草后面：起草多半会提出新的关系，开始时（_ProfileRun._rescope）按那时的目录再挑；现在挑不出也照建 - if not await _behind_draft(session, row.id): + if not allow_empty and not await _behind_draft(session, row.id): raise picked = [] + snapshot = settings.to_dict() + if trigger != "manual": + snapshot.update(trigger=trigger, fire_at=fire_at, schedule_id=schedule_id) job = CatalogProfileJob(id=new_id(), source_id=row.id, status="queued", scope=scope, - settings=settings.to_dict(), created_by=actor, retry_of=retry_of) + settings=snapshot, created_by=actor, retry_of=retry_of) tasks = _tasks_for(job.id, picked, utcnow()) session.add(job) try: @@ -760,6 +874,8 @@ class _ProfileRun(runtime.JobRun): self._table_started = 0.0 #: 开始时按此刻的目录一张表都挑不出来（_rescope）：为什么，记在任务的提醒里 self._scope_note: str | None = None + #: 这个任务是怎么建的（trigger_of）：取值变化的 detected_by 按它记 + self._trigger: dict[str, Any] = {"trigger": "manual", "fire_at": None, "schedule_id": None} # ---------------------------------------------------------------- 停止、出错时收尾 @@ -815,14 +931,17 @@ class _ProfileRun(runtime.JobRun): await self._write(self._rescope, row, source) async with SessionLocal() as session: job = await session.get(CatalogProfileJob, self.job_id) - pending = [(t.seq, t.table_name) for t in await task_rows(session, job.id) if t.status == "pending"] + scope = job.scope + pending = [(t.seq, t.table_name, *task_modes(scope, t.checks)) for t in await task_rows(session, job.id) + if t.status == "pending"] self.actor = job.created_by + self._trigger = trigger_of(job) try: self.cx = catalog_profile.make_context(source, settings, options=options, outside=outside) except ValueError as e: # 不支持的数据库类型 raise runtime.JobFailed(str(e)) from e hooks = TableHooks(progress=self._progress, query=self._queried, stop_text=self.stop_text) - for seq, name in pending: + for seq, name, regular, watched in pending: # 每张表开始前重读数据源：停用了、关掉了剖析就停下；遮罩按最新的设置 stop = await self._write(self._recheck) if stop is not None: @@ -837,7 +956,8 @@ class _ProfileRun(runtime.JobRun): hooks.recorded = recorded self._seq = seq try: - await catalog_profile.profile_table(self.cx, name, db=self._write, hooks=hooks, actor=self.actor) + await catalog_profile.profile_table(self.cx, name, db=self._write, hooks=hooks, actor=self.actor, + regular=regular, observe=watched) except SnapshotTampered as e: raise runtime.JobFailed(str(e)) from e if self.cx.runner.stopped: @@ -974,12 +1094,21 @@ class _ProfileRun(runtime.JobRun): row_estimate=result.row_estimate.to_dict() if result.row_estimate else None, findings=list(result.findings) or None, skipped=[s.to_dict() for s in result.skipped] or None, date_ranges=list(result.date_ranges) or None, query_log=list(result.query_log) or None, - summary=result.summary(), skipped_count=result.skipped_count, unfinished=result.unfinished, + summary=_summary_of(result), skipped_count=result.skipped_count, unfinished=result.unfinished, added=result.added, updated=result.updated, removed=result.removed, version=result.version, error=result.error, finished_at=utcnow())) 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: + # 取值观察：观察记录、比出的取值变化、撤销「已列全」都和这张表的结果、目录在同一个事务里 + 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) async def _finish(self, session: AsyncSession) -> None: """剖析完：万一还有没收尾的表（进度回调写库失败之类）记失败；有 failed、partial、cancelled 的表是 partial，否则 @@ -1005,7 +1134,8 @@ class ProfileJobs(runtime.JobManager): manager = ProfileJobs() -__all__ = ["INTERRUPTED_ERROR", "KEEP_JOBS", "MAX_TABLES", "NOTHING_AT_ALL", "NOTHING_NEEDED", "NothingToRetry", +__all__ = ["INTERRUPTED_ERROR", "KEEP_JOBS", "MAX_TABLES", "NOTHING_AT_ALL", "NOTHING_NEEDED", "NOTHING_TO_MAINTAIN", + "NOTHING_TO_OBSERVE", "NothingToRetry", "OBSERVING_SCOPES", "SCOPES", "task_modes", "trigger_of", "ProfileBusy", "ProfileFinished", "ProfileJobs", "ProfileNothing", "THEN_PROFILE", "active_job", "busy_message", "cancel", "create_job", "due_reason", "ensure_free", "job_brief", "job_dict", "job_out", "last_detail", "last_facts", "last_profiles", "outside_of", "latest_job", "load", "load_task", "manager", "prune", "public_last", diff --git a/backend/app/data/catalog_profile_observe.py b/backend/app/data/catalog_profile_observe.py new file mode 100644 index 00000000..bebeef03 --- /dev/null +++ b/backend/app/data/catalog_profile_observe.py @@ -0,0 +1,442 @@ +"""取值观察：剖析引擎的四类观察检查（docs/design/next-round/03-数据接入与目录维护.md B2）。 + +**为什么要有它。** 码值出现新取值、比例剧变、业务日期断档、某天没数据，以前都看不到：确认过的码值列不再取分布，大表只看 +物理顺序最老的前 10 万行，没有最新业务日期、按日行数、总行数的观察和历史；「已列全」过期以后 SQL 检查还照它提示。 + +**四类检查**（和关系、码值、日期检查一样是剖析的检查项：计入每张表的预算、止损，每条查询都留痕）： + +- rows：这张表的行数（剖析开头估行数那一条的结果，不另发查询）；口径 stats（数据库统计的估算值）/ count（数到上限）； +- codes_observe：码值列的取值分布——**所有**码值列都观察，包括人工确认过、标了已列全的（目录-D6）；只读，不改含义； +- freshness：业务日期列（已确认或已验证）的最新日期； +- daily_volume：业务日期最近 35 天每天的行数。 + +**大表的口径**（目录-D7）：小表（行数不超过整表统计的上限）看整表（full）。大表：业务日期已确认或已验证、而且这一列打头 +有索引（探查记下的 index_first_columns）时看最近 35 天的窗口（window，走索引的范围扫描）；PostgreSQL / Oracle 没有就块采样 +（sample，按数据块随机取约 p%，p 按表的大小取一档，不跨档比）；MySQL / SQLite 没有块采样，只看按存储顺序的前 N 行 +（first_rows），照实写口径。最新业务日期、按日行数在大表上只在业务日期列打头有索引时才查，否则记「表太大」跳过。每条观察 +都记口径（basis），只和口径相同的历史比（data/value_drift）。 + +**遮罩**：遮罩列不出现在任何一条观察查询里（被遮罩的码值列、业务日期列记「被遮罩」跳过；窗口要用的业务日期列被遮罩时 +不用窗口）。 + +**落库与比较**（save，和这张表的剖析结果、目录在同一个事务里）：每条观察写进 table_observations，每张表每种观察（码值按 +列）留最近 26 条；拿同口径的历史比出取值变化写进 source_changes。机器唯一能写进目录的事：数据证明码值表不再列全时撤销 +「已列全」（目录版本 +1，现有的目录漂移、冻结快照、SQL 检查自然跟上）；新取值本身、含义、状态都不改，交给人在变化里处理。 + +**什么时候观察**：剖析任务的范围 observe（只观察）和 maintenance（需要剖析的表 + 观察到期的表，定时维护用）才带观察检查； +人点的「需要剖析 / 全部 / 指定的表」照旧不观察。观察到期（observe_due）：有数据目录、不是视图的表，从没观察过，或者上一次 +观察在 OBSERVE_EVERY_H 小时以前。 +""" +from __future__ import annotations + +import logging +from collections.abc import Iterable, Mapping +from datetime import date, datetime, timedelta, timezone +from typing import Any + +from sqlalchemy import delete, func, select +from sqlalchemy.ext.asyncio import AsyncSession + +from app.data import catalog +from app.db.base import utcnow +from app.db.models_sources import TableObservation + +logger = logging.getLogger(__name__) + +OBSERVE_KINDS = ("rows", "codes_observe", "freshness", "daily_volume") +#: 按日行数、窗口观察看最近几天 +WINDOW_DAYS = 35 +#: 码值观察最多取几个取值（多出来的记 truncated，不判新码值、不算比例） +OBSERVE_CODES_MAX = 50 +#: 块采样的比例按表的大小取一档：目标约 max_scan_rows 行，向下取到这几档里的一档（同一张表大小不变时口径不变） +SAMPLE_LADDER = (10.0, 5.0, 2.0, 1.0, 0.5, 0.2, 0.1, 0.05, 0.02, 0.01) +#: 上一次观察多久以前算「观察到期」（每天跑一次的定时维护每次都会观察；每小时跑一次的一天观察两次） +OBSERVE_EVERY_H = 12 +#: 撤销「已列全」时目录的署名 +OBSERVER = "取值观察" + +KIND_LABEL = {"rows": "行数", "codes_observe": "码值观察", "freshness": "最新业务日期", "daily_volume": "按日行数"} + +TOO_LARGE_DATE = ("{table} 的行数超过整表统计的行数上限，业务日期列 {column} 打头没有索引，不看{what}" + "（大表只在业务日期列打头有索引时才查）") +NO_DATE = "{table} 没有已确认或已验证的业务日期，不看{what}" + + +# ========================================================================== +# 列出检查（纯函数） +# ========================================================================== + + +def _business_date(meta: Mapping[str, Any], notes: Mapping[str, Any]) -> str | None: + """已确认或已验证的业务日期列（表结构里的写法）；没有为 None。""" + from app.data.catalog_profile import _column + + item = notes.get("business_date") if isinstance(notes.get("business_date"), Mapping) else None + if not item or item.get("status") not in ("confirmed", "verified"): + return None + column = str((item.get("value") or {}).get("column") or "") + return _column(dict(meta), column) if column else None + + +def code_columns(meta: Mapping[str, Any], notes: Mapping[str, Any]) -> list[str]: + """要观察的码值列：目录里有码值项（没被驳回）的列，加上 ENUM / SET 列；按表结构的顺序。人工确认过、标了已列全的照样 + 观察（目录-D6）。""" + from app.data.catalog_profile import _is_enum + + described = notes.get("columns") if isinstance(notes.get("columns"), Mapping) else {} + out: list[str] = [] + for col in meta.get("columns") or []: + if not isinstance(col, Mapping): + continue + name = str(col.get("name") or "") + items = described.get(name) if isinstance(described.get(name), Mapping) else {} + codes = items.get("codes") if isinstance(items, Mapping) else None + listed = isinstance(codes, Mapping) and codes.get("status") != "rejected" + if listed or (_is_enum(col.get("type"), col.get("enum_values")) and not isinstance(codes, Mapping)): + out.append(name) + return out + + +def plan_observations(table: str, meta: Mapping[str, Any] | None, notes: Mapping[str, Any] | None, *, + masked: set[str]) -> list[Any]: + """这张表要做的观察检查（catalog_profile.PlannedCheck），接在关系、码值、日期检查后面：行数、各码值列、最新业务日期、 + 按日行数。被遮罩的列照样列出来、带 skip（不查，让人看得见为什么没有）。视图、表结构里没有的返回空列表。""" + from app.data.catalog_profile import PlannedCheck + + if not isinstance(meta, Mapping) or meta.get("is_view"): + return [] + notes = notes or {} + out = [PlannedCheck("rows", table, None)] + for column in code_columns(meta, notes): + skip = (("masked", f"{table}.{column} 在数据源设置中被遮罩，不观察取值", False) + if column.lower() in masked else None) + out.append(PlannedCheck("codes_observe", column, f"columns.{column}.codes", skip, column=column)) + day = _business_date(meta, notes) + for kind, what in (("freshness", "最新业务日期"), ("daily_volume", "按日行数")): + if day is None: + continue + skip = (("masked", f"{table}.{day} 在数据源设置中被遮罩，不看{what}", False) if day.lower() in masked else None) + out.append(PlannedCheck(kind, day, "business_date", skip, column=day)) + return out + + +# ========================================================================== +# 执行（在 catalog_profile._TableCx 上发查询：预算、止损、留痕都在那里） +# ========================================================================== + + +def sample_percent(rows: int | None, target: int) -> float | None: + """块采样取几档：约 target 行、向下取到 SAMPLE_LADDER 的一档；表不大于 target 时不采样（None）。""" + if not rows or rows <= target: + return None + want = 100.0 * target / rows + return next((p for p in SAMPLE_LADDER if p <= want), SAMPLE_LADDER[-1]) + + +def _today() -> date: + return datetime.now(timezone.utc).date() + + +def _since() -> date: + return _today() - timedelta(days=WINDOW_DAYS) + + +def _window_ok(cx: Any, meta: Mapping[str, Any], day: str | None) -> bool: + """大表能不能看窗口：业务日期列已确认或已验证、打头有索引、没被遮罩。""" + if not day or cx.is_masked(day): + return False + return day.lower() in {str(c).lower() for c in meta.get("index_first_columns") or []} + + +def _observation(kind: str, column: str, basis: dict[str, Any], data: dict[str, Any], refs: list[int]) -> dict[str, Any]: + key = basis_key(basis) + return {"kind": kind, "column": column, "basis": {**basis, "query_refs": refs}, "basis_key": key, "data": data} + + +def basis_key(basis: Mapping[str, Any]) -> str: + """口径的规范写法：只和它相同的历史比。""" + method = basis.get("method") + if method == "window": + return f"window:{basis.get('column')}:{basis.get('days')}" + if method == "sample": + return f"sample:{basis.get('percent'):g}" + if method == "first_rows": + return f"first_rows:{basis.get('rows')}" + return str(method or "") + + +def basis_text(basis: Mapping[str, Any]) -> str: + """口径的说法（跳过、留痕的说明里用）。""" + method = basis.get("method") + if method == "window": + return f"业务日期 {basis.get('column')} 最近 {basis.get('days')} 天" + if method == "sample": + return f"块采样约 {basis.get('percent'):g}%" + if method == "first_rows": + return f"按存储顺序取前 {basis.get('rows')} 行" + if method == "stats": + return "数据库统计的估算值" + if method == "count": + return "逐行数出" + return "整表" + + +async def execute(cx: Any, table: str, check: Any, result: Any) -> None: + """执行一项观察检查，观察到的追加进 result.observations。按规则不查的抛 catalog_profile._Skip。""" + from app.data.catalog_profile import _Skip + + meta = cx.tables[table] + start = len(result.query_log) + size = await cx.size_with_data(table, result) if check.kind != "rows" else await cx.size(table, result) + if check.kind == "rows": + basis: dict[str, Any] = {"method": size.method} + data = {"rows": size.rows, "at_least": size.at_least, "method": size.method} + if size.method == "unknown": + raise _Skip("no_data", f"{table} 的行数没能得到（统计信息和计数都没查成），不记行数观察") + result.observations.append(_observation("rows", "", basis, data, [])) + return + before = len(result.observations) + if check.kind == "codes_observe": + await _codes(cx, table, str(check.column), meta, size, result) + else: + await _dates(cx, table, check.kind, str(check.column), meta, size, result) + if len(result.observations) > before: + # 留痕：这条观察是哪几条查询得出的（这张表 query_log 里的下标） + result.observations[-1]["basis"]["query_refs"] = list(range(start, len(result.query_log))) + + +async def _codes(cx: Any, table: str, column: str, meta: Mapping[str, Any], size: Any, result: Any) -> None: + from app.data import catalog_profile as cp + + if cx.is_masked(column): + raise cp._Skip("masked", f"{table}.{column} 在数据源设置中被遮罩，不观察取值", carry=False) + d = cx.dialect + col_meta = cp._column_meta(dict(meta), column) or {} + t, c = d.table(dict(meta), table), d.column(dict(meta), column) + n = OBSERVE_CODES_MAX + 1 + day = _business_date(meta, getattr(cx, "table_notes", None) or {}) + if size.within(cx.settings.max_scan_rows): + basis: dict[str, Any] = {"method": "full"} + sql = d.value_counts_sql(t, c, n, scan_cap=None) + elif _window_ok(cx, meta, day): + since = _since() + basis = {"method": "window", "column": day, "days": WINDOW_DAYS, "since": since.isoformat()} + 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)) + 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 + else: + cap = cx.scan_cap(size) or max(cx.settings.sample_size, cx.settings.max_scan_rows) + basis, sql = {"method": "first_rows", "rows": cap}, d.value_counts_sql(t, c, n, scan_cap=cap) + res = await cx.run(sql, max_rows=n) + fixed = cp.is_fixed_char(col_meta.get("type")) + counts: dict[str, int] = {} + for row in res.rows[:OBSERVE_CODES_MAX]: + key = cp._code_key(row[0], fixed=fixed) + if key is None: + raise cp._Skip("unsupported", f"{table}.{column} 的取值不是整数或短文字，不观察", carry=False) + counts[key] = counts.get(key, 0) + (cp._as_int(row[1]) or 0) + data = {"values": [[k, v] for k, v in counts.items()], "rows": sum(counts.values()), + "truncated": len(res.rows) > OBSERVE_CODES_MAX} + result.observations.append(_observation("codes_observe", column, basis, data, [])) + + +async def _dates(cx: Any, table: str, kind: str, column: str, meta: Mapping[str, Any], size: Any, result: Any) -> None: + from app.data import catalog_profile as cp + + what = KIND_LABEL[kind] + if cx.is_masked(column): + raise cp._Skip("masked", f"{table}.{column} 在数据源设置中被遮罩，不看{what}", carry=False) + if not size.within(cx.settings.max_scan_rows) and not _window_ok(cx, meta, column): + raise cp._Skip("too_large", TOO_LARGE_DATE.format(table=table, column=column, what=what)) + d = cx.dialect + t, c = d.table(dict(meta), table), d.column(dict(meta), column) + text = catalog.column_family(cp._column_meta(dict(meta), column)) != "time" + if kind == "freshness": + res = await cx.run(d.max_date_sql(t, c, text=text)) + value = res.rows[0][0] if res.rows and res.rows[0] else None + if value is None or not cp._DATE_TEXT.match(str(value)): + raise cp._Skip("no_data", f"{table}.{column} 没有可识别的日期值") + result.observations.append(_observation("freshness", column, {"method": "full"}, + {"max": str(value)[:19]}, [])) + return + since = _since() + res = await cx.run(d.day_volume_sql(t, c, since, text=text), max_rows=WINDOW_DAYS + 2) + days: dict[str, int] = {} + for row in res.rows: + day = str(row[0])[:10] if row and row[0] is not None else "" + if cp._ISO_DAY.fullmatch(day): + days[day] = days.get(day, 0) + (cp._as_int(row[1]) or 0) + basis = {"method": "window", "column": column, "days": WINDOW_DAYS, "since": since.isoformat()} + result.observations.append(_observation("daily_volume", column, basis, + {"since": since.isoformat(), "until": _today().isoformat(), + "days": days}, [])) + + +# ========================================================================== +# 落库、比较、撤销「已列全」 +# ========================================================================== + + +async def _history(session: AsyncSession, source_id: str, table: str, kind: str, column: str) -> list[TableObservation]: + return list((await session.execute(select(TableObservation).where( + TableObservation.source_id == source_id, TableObservation.table_name == table, TableObservation.kind == kind, + TableObservation.column_name == column).order_by(TableObservation.at.desc(), TableObservation.id.desc()) + .limit(TableObservation.KEEP))).scalars()) + + +async def _prune(session: AsyncSession, source_id: str, table: str, kind: str, column: str) -> None: + keep = [r.id for r in await _history(session, source_id, table, kind, column)] + if keep: + await session.execute(delete(TableObservation).where( + TableObservation.source_id == source_id, TableObservation.table_name == table, + TableObservation.kind == kind, TableObservation.column_name == column, + TableObservation.id.not_in(keep)).execution_options(synchronize_session=False)) + + +def _rules_columns(notes: Mapping[str, Any], meta: Mapping[str, Any] | None, table: str, kind: str | None) -> set[str]: + """有效记录条件用到的列（小写）。""" + from app.data import sqlcheck + + item = notes.get("valid_filter") if isinstance(notes.get("valid_filter"), Mapping) else None + if not item or item.get("status") == "rejected": + return set() + parsed = sqlcheck.parse_valid_filter(item.get("value"), meta, kind=kind, table=table) + return {c.lower() for c in parsed.columns} + + +async def _revoke_complete(session: AsyncSession, source_id: str, table: str, column: str) -> int | None: + """撤销码值项上的「已列全」（目录版本 +1，不提交）：值、含义、状态、来源都不动。撞上别人刚改过重读重来，最多三次。 + 已经不是已列全了（别人刚改过）返回 None。""" + for _attempt in range(3): + entry = await catalog.read_entry(session, source_id, table) + if entry is None: + return None + notes = entry.notes + item = ((notes.get("columns") or {}).get(column) or {}).get("codes") + if not catalog.codes_complete(item): + return None + item.pop("complete", None) + item["updated_at"] = catalog.now_iso() + try: + written = await catalog.write_entry(session, source_id, table, notes, if_version=entry.version, + actor=OBSERVER, commit=False) + except catalog.CatalogConflict: + continue + return written.version + return None + + +async def save(session: AsyncSession, source: Any, table: str, result: Any, *, job_id: str | None, + seq: int | None, detected: Any) -> int: + """这张表这一次的观察写进 table_observations、和同口径的历史比，比出来的取值变化写进 source_changes（不提交：和剖析 + 结果、目录在同一个事务里）。返回记了几条新变化。出错只记日志、不挡剖析（观察是辅助步骤）。""" + from app.data import source_changes as sc + from app.data import value_drift as vd + + observations = list(getattr(result, "observations", None) or []) + if not observations: + return 0 + fresh = 0 + try: + entry = await catalog.read_entry(session, source.id, table) + notes = entry.notes if entry else {} + meta = ((getattr(source, "schema_cache", None) or {}).get("tables") or {}).get(table) + rules = _rules_columns(notes, meta, table, getattr(source, "kind", None)) + now = utcnow() + today = now.date() + for obs in observations: + kind, column = obs["kind"], obs["column"] + history = await _history(session, source.id, table, kind, column) + session.add(TableObservation(source_id=source.id, table_name=table, column_name=column, kind=kind, + basis=obs["basis"], basis_key=obs["basis_key"], data=obs["data"], + job_id=job_id, task_seq=seq, at=now)) + await session.flush() + await _prune(session, source.id, table, kind, column) + item = ((notes.get("columns") or {}).get(column) or {}).get("codes") if column else None + known = list((item or {}).get("value") or {}) if isinstance(item, Mapping) and \ + item.get("status") != "rejected" else [] + current = vd.Obs(kind, obs["basis_key"], obs["data"], now) + past = [vd.Obs(h.kind, h.basis_key, h.data or {}, h.at) for h in history] + findings = vd.compare(current, past, today=today, known=known, + complete=catalog.codes_complete(item)) + for f in findings: + after = dict(f.after) if isinstance(f.after, Mapping) else {"value": f.after} + if f.revoke_complete: + version = await _revoke_complete(session, source.id, table, column) + after.update(complete_revoked=version is not None, catalog_version=version) + touched = bool(column) and column.lower() in rules + change = sc.NewChange(kind=f.kind, table=table, column=column or None, severity=f.severity, + before=f.before, after=after, + basis={"method": obs["basis"].get("method"), "key": obs["basis_key"], + "query_refs": obs["basis"].get("query_refs") or []}, + affects={"catalog": [], "templates": [], "rules_touched": touched}, + key=(obs["basis_key"], f.key)) + _row, new = await sc.record(session, source, change, detected) + fresh += int(new) + except Exception: # noqa: BLE001 - 观察是辅助步骤：落库、比较出错不挡剖析结果 + logger.exception("保存 %s 的取值观察时出错", table) + return fresh + + +# ========================================================================== +# 观察到期、给助手的观察结论 +# ========================================================================== + + +async def last_observed(session: AsyncSession, source_id: str, + tables: Iterable[str] | None = None) -> dict[str, datetime]: + """每张表最近一次观察的时刻。""" + query = select(TableObservation.table_name, func.max(TableObservation.at)).where( + TableObservation.source_id == source_id).group_by(TableObservation.table_name) + if tables is not None: + wanted = list(tables) + if not wanted: + return {} + query = query.where(TableObservation.table_name.in_(wanted)) + return {name: at for name, at in (await session.execute(query)).all() if at is not None} + + +def observe_reason(last: datetime | None, *, now: datetime | None = None) -> str | None: + """观察到期没有：never 从没观察过；observe_due 上一次在 OBSERVE_EVERY_H 小时以前；没到期为 None。""" + if last is None: + return "never" + now = now or utcnow() + if last.tzinfo is None: + last = last.replace(tzinfo=timezone.utc) + return "observe_due" if now - last >= timedelta(hours=OBSERVE_EVERY_H) else None + + +async def observed_facts(session: AsyncSession, source_id: str, + tables: Iterable[str] | None = None) -> dict[str, dict[str, Any]]: + """给助手看的观察结论（经 schema_view.profile_lines 和剖析事实同一节，next-round X7）：{表名: {freshness: [{column, max, + at}], rows: {rows, method, at_least, at}}}，每种取最近一次。遮罩在渲染那一刻再核（不在这里滤）。""" + query = select(TableObservation).where(TableObservation.source_id == source_id, + TableObservation.kind.in_(("freshness", "rows"))) + if tables is not None: + wanted = list(tables) + if not wanted: + return {} + query = query.where(TableObservation.table_name.in_(wanted)) + out: dict[str, dict[str, Any]] = {} + seen: set[tuple[str, str, str]] = set() + for r in (await session.execute(query.order_by(TableObservation.at.desc()))).scalars(): + key = (r.table_name, r.kind, r.column_name) + if key in seen: + continue + seen.add(key) + facts = out.setdefault(r.table_name, {}) + at = r.at.isoformat() if r.at else None + if r.kind == "freshness" and (r.data or {}).get("max"): + facts.setdefault("freshness", []).append({"column": r.column_name, "max": r.data["max"], "at": at}) + elif r.kind == "rows": + facts["rows"] = {**{k: (r.data or {}).get(k) for k in ("rows", "method", "at_least")}, "at": at} + return out + + +__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"] diff --git a/backend/app/data/schema_view.py b/backend/app/data/schema_view.py index ba7af951..9f95bd66 100644 --- a/backend/app/data/schema_view.py +++ b/backend/app/data/schema_view.py @@ -65,6 +65,9 @@ PROFILE_ROWS_PROMPT = {"stats": "约 {n} 行（数据库统计估算）", "count PROFILE_RANGE_PROMPT = "{column} 从 {low} 到 {high}" PROFILE_HINT_PROMPT = "只供写查询参考（定时间范围、估表大小），不是查询结果，不能作为回答里的数" PROFILE_STALE_PROMPT = "数据剖析：剖析早于当前版本，不给" +#: 取值观察的结论（next-round 03 B2）：最新业务日期；行数比剖析的新时用观察的 +PROFILE_LATEST_PROMPT = "最新业务日期 {column} 到 {value}（取值观察，{day}）" +OBSERVE_HEAD_PROMPT = "取值观察（{day}）：{facts}" #: 三种「不在」（absent_detail）。{source} 数据源名，{table} 点名的表 ABSENT_REMOVED_PROMPT = "数据源「{source}」里已经没有 {table}（表已删除或改名）。别拿名字相近的表顶替" @@ -229,10 +232,15 @@ def profile_lines(facts: Mapping[str, Any] | None, masked: set[str], snapshot_id 之前一律不给。没有剖析事实（或者滤完什么都不剩）返回空列表：表详情和以前一字不差。""" if not facts: return [] + observed = facts.get("observed") if isinstance(facts.get("observed"), Mapping) and not upload else {} if upload and (not snapshot_id or facts.get("snapshot_id") != snapshot_id): return [indent + PROFILE_STALE_PROMPT] parts: list[str] = [] estimate = facts.get("row_estimate") if isinstance(facts.get("row_estimate"), Mapping) else None + seen_rows = observed.get("rows") if isinstance(observed.get("rows"), Mapping) else None + if seen_rows and str(seen_rows.get("at") or "") > str(facts.get("at") or ""): + # 取值观察记的行数比上次剖析的新：用新的（同一种说法，口径照写） + estimate = seen_rows if estimate: method, rows, least = estimate.get("method"), estimate.get("rows"), estimate.get("at_least") if isinstance(rows, int) and method in ("stats", "count"): @@ -246,10 +254,19 @@ def profile_lines(facts: Mapping[str, Any] | None, masked: set[str], snapshot_id if not column or column.lower() in masked or found.get("min") is None or found.get("max") is None: continue parts.append(PROFILE_RANGE_PROMPT.format(column=column, low=_when(found["min"]), high=_when(found["max"]))) + for found in observed.get("freshness") or []: + column = str((found or {}).get("column") or "") + if column and column.lower() not in masked and found.get("max"): + parts.append(PROFILE_LATEST_PROMPT.format(column=column, value=_when(found["max"]), day=_day(found.get("at")))) if not parts: return [] - return [indent + PROFILE_HEAD_PROMPT.format(day=_day(facts.get("at")), facts="；".join(parts)), - indent + " " + PROFILE_HINT_PROMPT] + if facts.get("at"): + head = PROFILE_HEAD_PROMPT.format(day=_day(facts.get("at")), facts="；".join(parts)) + else: + # 只有取值观察、没剖析过：照实说是观察 + latest = max((str(f.get("at") or "") for f in [seen_rows or {}, *(observed.get("freshness") or [])]), default="") + head = OBSERVE_HEAD_PROMPT.format(day=_day(latest), facts="；".join(parts)) + return [indent + head, indent + " " + PROFILE_HINT_PROMPT] # ========================================================================== diff --git a/backend/app/data/value_drift.py b/backend/app/data/value_drift.py new file mode 100644 index 00000000..a165f995 --- /dev/null +++ b/backend/app/data/value_drift.py @@ -0,0 +1,263 @@ +"""取值观察的比较器（docs/design/next-round/03-数据接入与目录维护.md B2 ④）：纯函数，不连库。 + +**为什么要有它。** 确认过的码值列以前不再取分布，新状态永远发现不了；大表只看最老的前 10 万行；没有最新业务日期、按日 +行数、总行数的历史。剖析引擎的四类观察检查（data/catalog_profile_observe）把每次看到的记进 table_observations，这里拿这一次 +和**同一口径**（basis_key 相同）的历史比，得出取值变化（写进 source_changes）。口径不同不比、照实说（comparable）。 + +规则（阈值第一版，要用真实库跑一段时间再调）： + +- **新码值**（codes_new）：码值表标了「已列全」的，出现任何一个目录里没有的取值都算——数据证明它不再列全，这是机器唯一能 + 写进目录的事（撤销「已列全」，revoke_complete），严重程度是「要处理」。没列全的，新值要占非空行的至少 0.5%，而且要有同 + 口径的历史（第一次只建基线），已见过的（目录里的、历史里出现过的）不算新。 +- **比例剧变**（codes_shift）：和上一次同口径的观察算 PSI，两次都至少 1,000 行、都没被截断；PSI ≥ 0.25 记 warning，0.1–0.25 + 记 info，列出变化最大的几个取值。 +- **断档**（date_stalled）：最新业务日期没前进，且这个日期离今天超过 max(2 × 以前前进的中位间隔, 3 天)；要见过至少一次 + 前进才知道它多久前进一次。 +- **缺天**（date_gap）：窗口里（有数据的第一天到最后一天之间）有日子没数据，而以前同口径的窗口都没有过。 +- **量跌**（volume_drop）：最近一个整天的行数不到它前面 28 天中位数的 30%（前 28 天至少 14 天有数据、中位数至少 20 行）。 + 最后一天和观察当天同一天（还在写入）时往前退一天。只看这一次的窗口，不需要历史。 +- **行数**（rows_drop / rows_jump）：比上一次同口径的少了 30% 以上 / 涨到 3 倍以上；依据是数据库统计的估算值时，要连续 + 两次都这样（和跌、涨之前那一次比）才报。数到上限、只知道「至少多少行」的不比。 +""" +from __future__ import annotations + +import math +import statistics +from collections.abc import Iterable, Mapping +from dataclasses import dataclass, field +from datetime import date, datetime, timedelta +from typing import Any + +NEW_CODE_SHARE = 0.005 +PSI_MIN_ROWS = 1000 +PSI_WARN = 0.25 +PSI_INFO = 0.1 +PSI_EPS = 1e-4 +PSI_TOP = 5 +STALL_MIN_DAYS = 3 +VOLUME_RATIO = 0.3 +VOLUME_BASELINE_DAYS = 28 +VOLUME_MIN_DAYS = 14 +VOLUME_MIN_MEDIAN = 20 +ROWS_DROP = 0.3 +ROWS_JUMP = 3.0 +GAPS_SHOWN = 10 + + +@dataclass(frozen=True) +class Obs: + """一次观察：kind、口径（basis_key）、观察到的（data）、什么时候（at）。""" + + kind: str + basis_key: str + data: Mapping[str, Any] + at: datetime | None = None + + +@dataclass(frozen=True) +class Finding: + """比出来的一条取值变化。key 是去重指纹的一部分；revoke_complete：要撤销码值项的「已列全」。""" + + kind: str + severity: str + before: Any = None + after: Any = None + key: Any = None + revoke_complete: bool = False + extra: dict[str, Any] = field(default_factory=dict) + + +def comparable(current: Obs, history: Iterable[Obs]) -> tuple[list[Obs], str | None]: + """同口径的历史（新的在前，照传进来的顺序）和一句说明：有历史、但口径都不同时说「口径不同，不比」；没有历史为 None。""" + items = list(history) + same = [h for h in items if h.basis_key == current.basis_key] + if items and not same: + return [], f"以前的观察口径不同（{items[0].basis_key} → {current.basis_key}），这一次只建新口径的基线、不比" + return same, None + + +def _counts(data: Mapping[str, Any]) -> dict[str, int]: + out: dict[str, int] = {} + for pair in data.get("values") or []: + if isinstance(pair, (list, tuple)) and len(pair) == 2 and isinstance(pair[1], (int, float)): + out[str(pair[0])] = out.get(str(pair[0]), 0) + int(pair[1]) + return out + + +def _share(n: int, total: int) -> float: + return round(n / total, 4) if total else 0.0 + + +def psi(before: Mapping[str, int], after: Mapping[str, int]) -> float: + """两个分布的 PSI（群体稳定性指数）：Σ (q − p) · ln(q / p)，p、q 是各取值占比，没出现的取值按 PSI_EPS 算。""" + total_b, total_a = sum(before.values()), sum(after.values()) + if not total_b or not total_a: + return 0.0 + value = 0.0 + for code in set(before) | set(after): + p = max(before.get(code, 0) / total_b, PSI_EPS) + q = max(after.get(code, 0) / total_a, PSI_EPS) + value += (q - p) * math.log(q / p) + return value + + +def codes(current: Obs, history: Iterable[Obs], *, known: Iterable[str] = (), complete: bool = False) -> list[Finding]: + """码值列的这一次和同口径历史比：新码值、比例剧变。known 是目录里码值项列出的取值；complete 是它标了「已列全」。""" + same, _note = comparable(current, history) + now = _counts(current.data) + rows = int(current.data.get("rows") or sum(now.values())) + listed = {str(k) for k in known} + out: list[Finding] = [] + if complete: + fresh = [c for c in now if c not in listed] + if fresh: + out.append(Finding("codes_new", "action", before={"known": sorted(listed), "complete": True}, + after={"codes": fresh, "shares": {c: _share(now[c], rows) for c in fresh}}, + key=tuple(sorted(fresh)), revoke_complete=True)) + elif same and not current.data.get("truncated") and not same[0].data.get("truncated"): + seen = listed | {c for h in same for c in _counts(h.data)} + fresh = [c for c, n in now.items() if c not in seen and rows and n / rows >= NEW_CODE_SHARE] + if fresh: + out.append(Finding("codes_new", "warning", before={"known": sorted(seen)}, + after={"codes": fresh, "shares": {c: _share(now[c], rows) for c in fresh}}, + key=tuple(sorted(fresh)))) + if same: + prev = same[0] + then = _counts(prev.data) + prev_rows = int(prev.data.get("rows") or sum(then.values())) + if (rows >= PSI_MIN_ROWS and prev_rows >= PSI_MIN_ROWS and not current.data.get("truncated") + and not prev.data.get("truncated")): + value = psi(then, now) + if value >= PSI_INFO: + top = sorted(set(then) | set(now), + key=lambda c: -abs(_share(now.get(c, 0), rows) - _share(then.get(c, 0), prev_rows))) + moved = [[c, _share(then.get(c, 0), prev_rows), _share(now.get(c, 0), rows)] for c in top[:PSI_TOP]] + out.append(Finding("codes_shift", "warning" if value >= PSI_WARN else "info", + before={"rows": prev_rows}, after={"psi": round(value, 4), "rows": rows, "top": moved}, + key=(current.basis_key, current.at.isoformat() if current.at else None))) + return out + + +def _day(value: Any) -> date | None: + try: + return date.fromisoformat(str(value)[:10]) + except ValueError: + return None + + +def freshness(current: Obs, history: Iterable[Obs], *, today: date) -> list[Finding]: + """最新业务日期没前进、而且离今天太久（断档）。""" + same, _note = comparable(current, history) + now = _day(current.data.get("max")) + if now is None or not same: + return [] + previous = [d for d in (_day(h.data.get("max")) for h in same) if d is not None] + if not previous or now > max(previous): + return [] + distinct = sorted(set(previous) | {now}) + gaps = [(b - a).days for a, b in zip(distinct, distinct[1:])] + if not gaps: + return [] + median = statistics.median(gaps) + threshold = max(2 * median, STALL_MIN_DAYS) + age = (today - now).days + if age <= threshold: + return [] + return [Finding("date_stalled", "warning", before={"max": max(previous).isoformat(), "median_days": median}, + after={"max": now.isoformat(), "age_days": age, "threshold_days": threshold}, + key=now.isoformat())] + + +def _series(data: Mapping[str, Any]) -> dict[date, int]: + out: dict[date, int] = {} + for raw, n in (data.get("days") or {}).items(): + d = _day(raw) + if d is not None and isinstance(n, (int, float)): + out[d] = out.get(d, 0) + int(n) + return out + + +def _holes(days: Mapping[date, int]) -> list[date]: + if not days: + return [] + first, last = min(days), max(days) + return [first + timedelta(days=i) for i in range((last - first).days + 1) + if days.get(first + timedelta(days=i), 0) == 0] + + +def daily(current: Obs, history: Iterable[Obs], *, today: date) -> list[Finding]: + """按日行数：缺天（要和同口径的历史比）、量跌（只看这一次的窗口）。""" + same, _note = comparable(current, history) + days = _series(current.data) + out: list[Finding] = [] + holes = _holes(days) + if holes and same and all(not _holes(_series(h.data)) for h in same): + out.append(Finding("date_gap", "warning", before={"gaps": 0}, + after={"gaps": len(holes), "days": [d.isoformat() for d in holes[:GAPS_SHOWN]]}, + key=tuple(d.isoformat() for d in holes))) + if days: + last = max(days) + if last >= today: + last -= timedelta(days=1) + base = [days.get(last - timedelta(days=i), 0) for i in range(1, VOLUME_BASELINE_DAYS + 1)] + present = sum(1 for n in base if n > 0) + median = statistics.median(base) if base else 0 + value = days.get(last, 0) + if present >= VOLUME_MIN_DAYS and median >= VOLUME_MIN_MEDIAN and value < VOLUME_RATIO * median: + out.append(Finding("volume_drop", "warning", before={"median": median, "days": VOLUME_BASELINE_DAYS}, + after={"day": last.isoformat(), "rows": value, "ratio": round(value / median, 4)}, + key=last.isoformat())) + return out + + +def _exact(data: Mapping[str, Any]) -> int | None: + rows = data.get("rows") + return int(rows) if isinstance(rows, int) and not isinstance(rows, bool) and rows >= 0 else None + + +def rows(current: Obs, history: Iterable[Obs]) -> list[Finding]: + """行数：跌 30% 以上、涨到 3 倍以上。统计估算的要连续两次（都和跌、涨之前那一次比）。""" + same, _note = comparable(current, history) + now = _exact(current.data) + if now is None or not same: + return [] + estimated = current.data.get("method") == "stats" + if estimated: + if len(same) < 2: + return [] + prev, ref = _exact(same[0].data), _exact(same[1].data) + if prev is None or ref is None or not ref: + return [] + dropped = now <= (1 - ROWS_DROP) * ref and prev <= (1 - ROWS_DROP) * ref + jumped = now >= ROWS_JUMP * ref and prev >= ROWS_JUMP * ref + else: + ref = _exact(same[0].data) + if ref is None or not ref: + return [] + dropped = now <= (1 - ROWS_DROP) * ref + jumped = now >= ROWS_JUMP * ref + if dropped: + return [Finding("rows_drop", "warning", before={"rows": ref}, after={"rows": now, "estimated": estimated}, + key=(current.basis_key, ref))] + if jumped: + return [Finding("rows_jump", "info", before={"rows": ref}, after={"rows": now, "estimated": estimated}, + key=(current.basis_key, ref))] + return [] + + +def compare(current: Obs, history: Iterable[Obs], *, today: date, known: Iterable[str] = (), + complete: bool = False) -> list[Finding]: + """按种类分派。""" + if current.kind == "codes_observe": + return codes(current, history, known=known, complete=complete) + if current.kind == "freshness": + return freshness(current, history, today=today) + if current.kind == "daily_volume": + return daily(current, history, today=today) + if current.kind == "rows": + return rows(current, history) + return [] + + +__all__ = ["Finding", "NEW_CODE_SHARE", "Obs", "PSI_INFO", "PSI_MIN_ROWS", "PSI_WARN", "codes", "comparable", + "compare", "daily", "freshness", "psi", "rows"] diff --git a/backend/tests/fixtures/dialect/mariadb.sql b/backend/tests/fixtures/dialect/mariadb.sql index 0130c4d9..b837c0b0 100644 --- a/backend/tests/fixtures/dialect/mariadb.sql +++ b/backend/tests/fixtures/dialect/mariadb.sql @@ -25,3 +25,7 @@ SELECT status, COUNT(*) AS cnt FROM sales.orders WHERE paid_at >= DATE '2026-08- 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 -- （这种方言没有这类查询） +-- max_date +SELECT MAX(paid_at) FROM sales.orders WHERE paid_at >= DATE '0001-01-02' AND paid_at < DATE '9999-12-31'; +-- max_date_text +SELECT MAX(biz_day) FROM sales.orders WHERE biz_day >= '0001-01-02' AND biz_day < '9999-12-31'; diff --git a/backend/tests/fixtures/dialect/mysql.sql b/backend/tests/fixtures/dialect/mysql.sql index 1b046e36..17f0f5da 100644 --- a/backend/tests/fixtures/dialect/mysql.sql +++ b/backend/tests/fixtures/dialect/mysql.sql @@ -25,3 +25,7 @@ SELECT status, COUNT(*) AS cnt FROM sales.orders WHERE paid_at >= DATE '2026-08- 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 -- （这种方言没有这类查询） +-- max_date +SELECT MAX(paid_at) FROM sales.orders WHERE paid_at >= DATE '0001-01-02' AND paid_at < DATE '9999-12-31'; +-- max_date_text +SELECT MAX(biz_day) FROM sales.orders WHERE biz_day >= '0001-01-02' AND biz_day < '9999-12-31'; diff --git a/backend/tests/fixtures/dialect/oracle.sql b/backend/tests/fixtures/dialect/oracle.sql index c338ead9..ba09c3d6 100644 --- a/backend/tests/fixtures/dialect/oracle.sql +++ b/backend/tests/fixtures/dialect/oracle.sql @@ -25,3 +25,7 @@ SELECT * FROM (SELECT status, COUNT(*) AS cnt FROM sales.orders WHERE paid_at >= 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; +-- max_date +SELECT MAX(paid_at) FROM sales.orders WHERE paid_at >= DATE '0001-01-02' AND paid_at < DATE '9999-12-31'; +-- max_date_text +SELECT MAX(biz_day) FROM sales.orders WHERE biz_day >= '0001-01-02' AND biz_day < '9999-12-31'; diff --git a/backend/tests/fixtures/dialect/postgresql.sql b/backend/tests/fixtures/dialect/postgresql.sql index 1416da24..995eb3f3 100644 --- a/backend/tests/fixtures/dialect/postgresql.sql +++ b/backend/tests/fixtures/dialect/postgresql.sql @@ -25,3 +25,7 @@ SELECT status, COUNT(*) AS cnt FROM sales.orders WHERE paid_at >= DATE '2026-08- 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; +-- max_date +SELECT MAX(paid_at) FROM sales.orders WHERE paid_at >= DATE '0001-01-02' AND paid_at < DATE '9999-12-31'; +-- max_date_text +SELECT MAX(biz_day) FROM sales.orders WHERE biz_day >= '0001-01-02' AND biz_day < '9999-12-31'; diff --git a/backend/tests/fixtures/dialect/sqlite.sql b/backend/tests/fixtures/dialect/sqlite.sql index 8b10405f..44e0a904 100644 --- a/backend/tests/fixtures/dialect/sqlite.sql +++ b/backend/tests/fixtures/dialect/sqlite.sql @@ -25,3 +25,7 @@ SELECT status, COUNT(*) AS cnt FROM sales.orders WHERE paid_at >= '2026-08-27' A 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 -- （这种方言没有这类查询） +-- max_date +SELECT MAX(paid_at) FROM sales.orders WHERE paid_at >= '0001-01-02' AND paid_at < '9999-12-31'; +-- max_date_text +SELECT MAX(biz_day) FROM sales.orders WHERE biz_day >= '0001-01-02' AND biz_day < '9999-12-31'; diff --git a/backend/tests/test_catalog_observe.py b/backend/tests/test_catalog_observe.py new file mode 100644 index 00000000..cd243cdd --- /dev/null +++ b/backend/tests/test_catalog_observe.py @@ -0,0 +1,291 @@ +"""取值观察：剖析引擎的四类观察检查、口径、撤销「已列全」、遮罩（next-round 03 B2）。 + +SQLite 夹具（虚构的店铺订单表：状态码、支付日、金额；支付日打头有索引）走真实的后台剖析任务（范围 observe）： + +- 四类观察各记一条，口径照实（小表整表；大表有索引的业务日期走 35 天窗口，没有就按存储顺序的前 N 行）； +- 确认过、标了已列全的码值列也观察；插进一个新取值后再观察：记一条 codes_new（要处理）、撤销已列全（目录版本 +1，署名 + 「取值观察」，值、含义、状态都不动），SQL 检查不再按旧码值表提示 unknown_code； +- 口径不同不比；遮罩列不出现在任何一条观察查询里（看留痕）； +- PostgreSQL / Oracle 的大表没有可用的窗口时生成块采样 SQL，MySQL 退回前 N 行（假的执行上下文，只看拼出来的 SQL 和口径）。 +""" +from __future__ import annotations + +import sqlite3 +import uuid +from datetime import date, timedelta +from types import SimpleNamespace + +import pytest +from httpx import ASGITransport, AsyncClient +from sqlalchemy import select, update + +from app.data import catalog, catalog_profile, catalog_profile_jobs, catalog_profile_observe as observe, sqlcheck +from app.data import table_versions +from app.data.engine import MASK_COLUMNS_OPTION +from app.db.base import SessionLocal +from app.db.models import CatalogProfileTask, DataSource +from app.db.models_sources import SourceChange, TableObservation +from app.main import app +from tests.fixtures.sources import drop_source, introspect_now + +TODAY = date.today() + + +def _build(path, *, rows: int = 60, index: bool = True) -> None: + db = sqlite3.connect(path) + db.execute("CREATE TABLE orders (id INTEGER PRIMARY KEY, status VARCHAR(10), paid_on DATE, amount NUMERIC(10, 2)," + " secret VARCHAR(10))") + if index: + db.execute("CREATE INDEX ix_orders_paid_on ON orders (paid_on)") + for i in range(1, rows + 1): + day = (TODAY - timedelta(days=i % 30 + 1)).isoformat() + db.execute("INSERT INTO orders VALUES (?, ?, ?, ?, ?)", (i, "void" if i % 4 == 0 else "paid", day, i, "x")) + db.commit() + db.close() + + +@pytest.fixture +async def client(): + async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c: + yield c + + +async def _make(client, path, *, max_scan_rows: int = 100_000, masked: list[str] | None = None) -> str: + options = {"catalog_profile": {"enabled": True, "max_scan_rows": max_scan_rows, "sample_size": 10}} + if masked: + options[MASK_COLUMNS_OPTION] = masked + r = await client.post("/api/datasources", json={"name": f"obs_{uuid.uuid4().hex[:8]}", "kind": "sqlite", + "database": str(path), "readonly": True}) + assert r.status_code == 201, r.text + sid = r.json()["id"] + await introspect_now(client, sid) + async with SessionLocal() as session: + row = await session.get(DataSource, sid) + row.options = {**(row.options or {}), **options} + await session.commit() + await catalog.write_entry(session, sid, "orders", { + "business_date": catalog.make_item({"column": "paid_on"}, "human"), + "columns": {"status": {"codes": catalog.make_item({"paid": "已支付", "void": "作废"}, "human", + complete=True)}, + "secret": {"codes": catalog.make_item({"x": "保密"}, "human")}}, + "valid_filter": catalog.make_item("status <> 'void'", "human"), + }, if_version=0, actor="测试员") + return sid + + +@pytest.fixture +async def shop(client, tmp_path): + path = tmp_path / "obs.db" + _build(path) + sid = await _make(client, path) + yield sid, path + await drop_source(sid) + + +async def _observe(sid: str, scope: str = "observe") -> str: + async with SessionLocal() as session: + row = await session.get(DataSource, sid) + view = await table_versions.resolve_source(session, row) + job, _ = await catalog_profile_jobs.create_job(session, row, view, tables=None, scope=scope, + settings=catalog_profile.ensure_enabled(row), actor="测试员") + catalog_profile_jobs.manager.start(job.id) + await catalog_profile_jobs.manager.wait(job.id, timeout=60) + return job.id + + +async def _obs(sid: str) -> list[TableObservation]: + async with SessionLocal() as session: + return list((await session.execute(select(TableObservation).where(TableObservation.source_id == sid) + .order_by(TableObservation.at))).scalars()) + + +async def _age(sid: str, hours: int = 13) -> None: + """把以前的观察挪到 hours 小时前：观察到期了才会再观察。""" + async with SessionLocal() as session: + for r in (await session.execute(select(TableObservation).where(TableObservation.source_id == sid))).scalars(): + r.at = r.at - timedelta(hours=hours) + await session.commit() + + +async def test_four_observations_with_basis_and_no_change_on_first_run(shop): + sid, _ = shop + job_id = await _observe(sid) + got = {(o.kind, o.column_name): o for o in await _obs(sid)} + assert set(got) == {("rows", ""), ("codes_observe", "status"), ("codes_observe", "secret"), + ("freshness", "paid_on"), ("daily_volume", "paid_on")} + assert got[("rows", "")].data["rows"] == 60 and got[("rows", "")].basis_key == "count" + status = got[("codes_observe", "status")] + assert status.basis_key == "full" and dict(map(tuple, status.data["values"])) == {"paid": 45, "void": 15} + assert got[("freshness", "paid_on")].data["max"] == (TODAY - timedelta(days=1)).isoformat() + days = got[("daily_volume", "paid_on")].data["days"] + assert sum(days.values()) == 60 and got[("daily_volume", "paid_on")].basis_key == "window:paid_on:35" + assert all(o.job_id == job_id for o in got.values()) + assert got[("codes_observe", "status")].basis["query_refs"], "留痕：这条观察是哪几条查询得出的" + async with SessionLocal() as session: + changes = list((await session.execute(select(SourceChange).where(SourceChange.source_id == sid))).scalars()) + [task] = (await session.execute(select(CatalogProfileTask).where(CatalogProfileTask.job_id == job_id))).scalars() + assert changes == [], "第一次只建基线" + assert {c["kind"] for c in task.checks} == set(observe.OBSERVE_KINDS), "只观察：不做关系、码值、日期的剖析" + assert task.status == "ok" + + +async def test_new_code_in_a_complete_list_revokes_complete(shop): + sid, path = shop + await _observe(sid) + db = sqlite3.connect(path) + db.execute("INSERT INTO orders VALUES (999, 'held', ?, 1, 'x')", ((TODAY - timedelta(days=2)).isoformat(),)) + db.commit() + db.close() + async with SessionLocal() as session: + before = await catalog.read_entry(session, sid, "orders") + # 没到期不再观察 + with pytest.raises(catalog_profile_jobs.ProfileNothing): + await _observe(sid) + await _age(sid) + await _observe(sid) + async with SessionLocal() as session: + [change] = (await session.execute(select(SourceChange).where( + SourceChange.source_id == sid, SourceChange.kind == "codes_new"))).scalars() + after = await catalog.read_entry(session, sid, "orders") + row = await session.get(DataSource, sid) + assert change.column_name == "status" and change.severity == "action" and change.after["codes"] == ["held"] + assert change.after["complete_revoked"] is True and change.after["catalog_version"] == after.version + assert change.affects["rules_touched"] is True, "有效记录条件用到了这一列" + assert change.detected_by["job_kind"] == "profile" and change.basis["key"] == "full" + assert after.version == before.version + 1 and after.updated_by == observe.OBSERVER + item = after.notes["columns"]["status"]["codes"] + assert "complete" not in item and item["value"] == {"paid": "已支付", "void": "作废"}, "值、含义不动" + assert (item["source"], item["status"]) == ("human", "confirmed"), "状态、来源不动" + checker = sqlcheck.SqlChecker(kind="sqlite", schema_cache=row.schema_cache, notes={"orders": after.notes}, + source_name=row.name) + assert not [c for c in checker.check("SELECT COUNT(*) FROM orders WHERE status = 'held'") + if c.code == "unknown_code"], "已列全撤销以后，SQL 检查不再照旧码值表提示" + old = sqlcheck.SqlChecker(kind="sqlite", schema_cache=row.schema_cache, notes={"orders": before.notes}, + source_name=row.name) + assert [c for c in old.check("SELECT COUNT(*) FROM orders WHERE status = 'held'") if c.code == "unknown_code"] + + # 同一个新码值再测到：不另起一条 + await _age(sid) + await _observe(sid) + async with SessionLocal() as session: + same = list((await session.execute(select(SourceChange).where( + SourceChange.source_id == sid, SourceChange.kind == "codes_new"))).scalars()) + assert len(same) == 1 + + +async def test_masked_columns_never_appear_in_observation_queries(client, tmp_path): + path = tmp_path / "masked.db" + _build(path) + sid = await _make(client, path, masked=["secret"]) + try: + job_id = await _observe(sid) + async with SessionLocal() as session: + [task] = (await session.execute(select(CatalogProfileTask).where( + CatalogProfileTask.job_id == job_id))).scalars() + assert task.query_log and not any("secret" in q["sql"].lower() for q in task.query_log) + [skip] = [c for c in task.checks if c["kind"] == "codes_observe" and c["target"] == "secret"] + assert skip["state"] == "skipped" and skip["reason"] == "masked" + assert ("codes_observe", "secret") not in {(o.kind, o.column_name) for o in await _obs(sid)} + finally: + await drop_source(sid) + + +async def test_large_table_basis_window_or_first_rows_and_no_cross_basis_compare(client, tmp_path): + with_index, without = tmp_path / "big1.db", tmp_path / "big2.db" + _build(with_index, rows=60, index=True) + _build(without, rows=60, index=False) + a = await _make(client, with_index, max_scan_rows=20) + b = await _make(client, without, max_scan_rows=20) + try: + await _observe(a) + await _observe(b) + got_a = {(o.kind, o.column_name): o for o in await _obs(a)} + got_b = {(o.kind, o.column_name): o for o in await _obs(b)} + assert got_a[("codes_observe", "status")].basis_key == "window:paid_on:35", "大表、业务日期打头有索引：窗口" + assert got_a[("rows", "")].data["at_least"] == 21 + assert got_b[("codes_observe", "status")].basis_key.startswith("first_rows:"), \ + "SQLite 没有块采样：按存储顺序的前 N 行，照实写口径" + assert ("freshness", "paid_on") not in got_b and ("daily_volume", "paid_on") not in got_b, \ + "大表、业务日期没有索引：不看最新业务日期和按日行数" + # 改了口径（整表统计上限调大，变成小表）：和以前的不比 + async with SessionLocal() as session: + row = await session.get(DataSource, b) + row.options = {**row.options, "catalog_profile": {**row.options["catalog_profile"], "max_scan_rows": 1000}} + await session.commit() + db = sqlite3.connect(without) + db.executemany("INSERT INTO orders VALUES (?, 'void', ?, 1, 'x')", + [(1000 + i, (TODAY - timedelta(days=1)).isoformat()) for i in range(40)]) + db.commit() + db.close() + await _age(b) + await _observe(b) + async with SessionLocal() as session: + shifts = list((await session.execute(select(SourceChange).where( + SourceChange.source_id == b, SourceChange.kind == "codes_shift"))).scalars()) + assert shifts == [], "口径从前 N 行换成了整表：不跨口径比" + finally: + await drop_source(a) + await drop_source(b) + + +class _FakeCx: + """只看拼出来的 SQL：大表、业务日期没有索引。""" + + def __init__(self, kind: str) -> None: + self.dialect = catalog_profile.SqlDialect(kind) + self.settings = catalog_profile.ProfileSettings(enabled=True, max_scan_rows=1000, sample_size=100) + self.sql: list[str] = [] + self.table_notes = {"business_date": {"value": {"column": "paid_on"}, "source": "human", "status": "confirmed"}} + + def is_masked(self, column: str) -> bool: + return False + + def scan_cap(self, size): + return 1000 + + async def run(self, sql, *, max_rows=1, **kw): + self.sql.append(sql) + return SimpleNamespace(rows=[("A", 900), ("B", 100)], row_count=2) + + +@pytest.mark.parametrize("kind,method", [("postgresql", "sample"), ("oracle", "sample"), ("mysql", "first_rows")]) +async def test_large_table_without_window_samples_on_pg_and_oracle(kind, method): + cx = _FakeCx(kind) + meta = {"columns": [{"name": "status", "type": "VARCHAR(2)"}, {"name": "paid_on", "type": "DATE"}], + "index_first_columns": ["id"]} + result = SimpleNamespace(observations=[], query_log=[]) + size = catalog_profile.TableSize(500_000, "stats") + await observe._codes(cx, "orders", "status", meta, size, result) + [obs] = result.observations + assert obs["basis"]["method"] == method + if kind == "postgresql": + assert "TABLESAMPLE SYSTEM (0.2)" in cx.sql[0] and obs["basis_key"] == "sample:0.2" + elif kind == "oracle": + assert "SAMPLE BLOCK (0.2)" in cx.sql[0] and "ROWNUM" in cx.sql[0] + else: + assert "LIMIT 1000" in cx.sql[0] and obs["basis_key"] == "first_rows:1000" + assert observe.sample_percent(500_000, 1000) == 0.2 and observe.sample_percent(800, 1000) is None + + +async def test_observation_is_not_the_last_profile_and_reaches_the_table_detail(shop): + """只做了观察的那一项不算「上次剖析」（不顶掉剖析事实）；观察结论经 schema_view 和剖析事实同一节进表详情（X7）， + 遮罩在渲染那一刻再核。""" + from app.data import schema_view + + sid, _ = shop + await _observe(sid) + async with SessionLocal() as session: + lasts = await catalog_profile_jobs.last_profiles(session, sid) + facts = await catalog_profile_jobs.last_facts(session, sid) + row = await session.get(DataSource, sid) + assert "orders" not in lasts, "只观察的那一项不是一次剖析" + observed = facts["orders"]["observed"] + assert observed["freshness"][0]["column"] == "paid_on" and observed["rows"]["rows"] == 60 + lines = schema_view.profile_lines(facts["orders"], set(), None) + latest = (TODAY - timedelta(days=1)).isoformat() + assert lines[0].strip().startswith("取值观察（") and f"最新业务日期 paid_on 到 {latest}" in lines[0], lines + assert "60 行（逐行数出）" in lines[0] and "不能作为回答里的数" in lines[1] + assert not any("paid_on" in line for line in schema_view.profile_lines(facts["orders"], {"paid_on"}, None)), \ + "观察之后才加遮罩的列：渲染那一刻滤掉" + detail = schema_view.table_detail(row, "orders", profile=facts) + assert f"最新业务日期 paid_on 到 {latest}" in detail diff --git a/backend/tests/test_profile_dialect_golden.py b/backend/tests/test_profile_dialect_golden.py index 7e7c6321..ebdd546e 100644 --- a/backend/tests/test_profile_dialect_golden.py +++ b/backend/tests/test_profile_dialect_golden.py @@ -3,7 +3,7 @@ 三层，各挡各的： 1. **金样**：每种方言（mysql、mariadb、postgresql、oracle、sqlite）× 每类查询（统计信息、有上限的计数、取不同键值、 - 对键值、唯一性、取值分布、日期最小最大、时间窗口里的取值分布、按日行数、块采样），完整 SQL 钉在 + 对键值、唯一性、取值分布、日期最小最大、时间窗口里的取值分布、按日行数、块采样、最新业务日期），完整 SQL 钉在 tests/fixtures/dialect/<方言>.sql，人审过一次；以后拼法有任何改动都会出 diff。要更新金样： ``UPDATE_DIALECT_GOLDEN=1 pytest tests/test_profile_dialect_golden.py``，再逐行看 git diff。 2. **sqlglot 逐条解析**：按方言读一遍（mariadb 按 mysql 读），挡住语法写坏了（少括号、关键字拼错）。 @@ -71,6 +71,9 @@ def _queries(kind: str) -> list[tuple[str, str | None]]: "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)), + # 第 3 波：取值观察的最新业务日期（文字存的日期按字符串比） + ("max_date", d.max_date_sql(orders, d.column(META["orders"], "paid_at"))), + ("max_date_text", d.max_date_sql(orders, d.column(META["orders"], "biz_day"), text=True)), ] return out diff --git a/backend/tests/test_value_drift.py b/backend/tests/test_value_drift.py new file mode 100644 index 00000000..6fa80ef9 --- /dev/null +++ b/backend/tests/test_value_drift.py @@ -0,0 +1,114 @@ +"""取值观察的比较器（app/data/value_drift.py，next-round 03 B2 ④）：纯函数。 + +守的是阈值两侧：已列全的码值表出现任何新值都算、要撤销「已列全」；没列全的新值要占 0.5% 以上、第一次只建基线；PSI 刚好落在 +0.1 / 0.25 两侧；断档要超过 max(2 × 中位间隔, 3 天)；缺天要以前的窗口都没有过；量跌按前 28 天的中位数；行数估算要连续两次； +口径不同不比。取值都是虚构的。 +""" +from __future__ import annotations + +from datetime import date, datetime, timedelta, timezone + +from app.data import value_drift as vd + +AT = datetime(2026, 10, 10, 3, 0, tzinfo=timezone.utc) +TODAY = date(2026, 10, 10) + + +def _codes(values: dict[str, int], basis: str = "full", **extra) -> vd.Obs: + return vd.Obs("codes_observe", basis, {"values": [[k, v] for k, v in values.items()], + "rows": sum(values.values()), **extra}, AT) + + +def test_complete_code_table_flags_any_new_value_and_revokes_complete(): + [f] = vd.codes(_codes({"paid": 900, "void": 99, "held": 1}), [], known=["paid", "void"], complete=True) + assert f.kind == "codes_new" and f.severity == "action" and f.revoke_complete is True + assert f.after["codes"] == ["held"], "只出现一次也算：它说的是「只有这些取值」" + assert vd.codes(_codes({"paid": 900, "void": 100}), [], known=["paid", "void"], complete=True) == [] + + +def test_not_complete_needs_history_and_half_a_percent(): + base = _codes({"paid": 900, "void": 100}) + assert vd.codes(_codes({"paid": 900, "void": 95, "held": 5}), [], known=[]) == [], "第一次只建基线" + [f] = vd.codes(_codes({"paid": 900, "void": 95, "held": 5}), [base], known=[]) + assert f.kind == "codes_new" and f.severity == "warning" and not f.revoke_complete + assert vd.codes(_codes({"paid": 996, "void": 100, "held": 4}), [base], known=[]) == [], "不到 0.5%：不算" + seen = _codes({"paid": 900, "void": 90, "held": 10}) + assert [x.kind for x in vd.codes(_codes({"paid": 900, "void": 95, "held": 5}), [seen, base])] == [], "历史里见过的不算新" + + +def test_psi_on_both_sides_of_the_thresholds(): + before = _codes({"a": 500, "b": 500}) + # PSI(0.5/0.5 → 0.6/0.4) ≈ 0.0405：不报 + assert vd.codes(_codes({"a": 600, "b": 400}), [before]) == [] + # 0.5/0.5 → 0.66/0.34 ≈ 0.1035：info + [info] = vd.codes(_codes({"a": 660, "b": 340}), [before]) + assert info.kind == "codes_shift" and info.severity == "info" and 0.1 <= info.after["psi"] < 0.25 + # 0.5/0.5 → 0.75/0.25 ≈ 0.2747：warning，列出变化最大的取值 + [warn] = vd.codes(_codes({"a": 750, "b": 250}), [before]) + assert warn.severity == "warning" and warn.after["psi"] >= 0.25 and warn.after["top"][0][0] in ("a", "b") + small = _codes({"a": 50, "b": 50}) + assert vd.codes(_codes({"a": 90, "b": 10}), [small]) == [], "两次都要至少 1,000 行" + + +def test_different_basis_is_not_compared(): + before = _codes({"a": 500, "b": 500}, basis="first_rows:1000") + assert vd.codes(_codes({"a": 900, "b": 100}, basis="window:paid_on:35"), [before]) == [] + same, note = vd.comparable(_codes({}, basis="window:paid_on:35"), [before]) + assert same == [] and "口径不同" in note + + +def _fresh(day: str) -> vd.Obs: + return vd.Obs("freshness", "full", {"max": day}, AT) + + +def test_stalled_date_needs_more_than_twice_the_median_interval_and_three_days(): + history = [_fresh("2026-10-05"), _fresh("2026-10-04"), _fresh("2026-10-03")] # 每天前进一次 + assert vd.freshness(_fresh("2026-10-05"), history, today=date(2026, 10, 8)) == [], "3 天：没超过 max(2, 3)" + [f] = vd.freshness(_fresh("2026-10-05"), history, today=date(2026, 10, 9)) + assert f.kind == "date_stalled" and f.after["age_days"] == 4 and f.after["threshold_days"] == 3 + assert vd.freshness(_fresh("2026-10-08"), history, today=date(2026, 10, 30)) == [], "前进了就不算断档" + weekly = [_fresh("2026-09-28"), _fresh("2026-09-21")] + assert vd.freshness(_fresh("2026-09-28"), weekly, today=date(2026, 10, 10)) == [], "每周一次：14 天内不算" + assert vd.freshness(_fresh("2026-10-05"), [_fresh("2026-10-05")], today=date(2026, 10, 30)) == [], \ + "没见过前进，不知道它多久前进一次" + + +def _days(start: date, counts: list[int], basis: str = "window:paid_on:35") -> vd.Obs: + return vd.Obs("daily_volume", basis, {"days": {(start + timedelta(days=i)).isoformat(): n + for i, n in enumerate(counts) if n}}, AT) + + +def test_gap_only_when_previous_windows_had_none(): + full = _days(date(2026, 9, 1), [100] * 30) + holed = _days(date(2026, 9, 2), [100] * 10 + [0, 0] + [100] * 18) + [f] = vd.daily(holed, [full], today=TODAY) + assert f.kind == "date_gap" and f.after["gaps"] == 2 + assert not [x for x in vd.daily(holed, [holed], today=TODAY) if x.kind == "date_gap"], "以前就有缺天：不算新问题" + assert not [x for x in vd.daily(holed, [], today=TODAY) if x.kind == "date_gap"], "没有历史：只建基线" + + +def test_volume_drop_against_the_previous_28_days(): + start = TODAY - timedelta(days=31) + normal = _days(start, [100] * 30) # 最后一天是昨天 + assert vd.daily(normal, [], today=TODAY) == [] + [f] = vd.daily(_days(start, [100] * 29 + [20]), [], today=TODAY) + assert f.kind == "volume_drop" and f.after["rows"] == 20 and f.before["median"] == 100 + assert vd.daily(_days(start, [100] * 29 + [35]), [], today=TODAY) == [], "35% 不算跌" + partial = _days(TODAY - timedelta(days=30), [100] * 30 + [5]) # 今天还在写入：往前退一天 + assert vd.daily(partial, [], today=TODAY) == [] + + +def _rows(n: int, method: str = "count") -> vd.Obs: + return vd.Obs("rows", method, {"rows": n, "method": method}, AT) + + +def test_rows_drop_and_jump_estimates_need_two_in_a_row(): + assert [f.kind for f in vd.rows(_rows(600), [_rows(1000)])] == ["rows_drop"] + assert vd.rows(_rows(750), [_rows(1000)]) == [], "跌 25%：不报" + assert [f.kind for f in vd.rows(_rows(3000), [_rows(1000)])] == ["rows_jump"] + est = lambda n: _rows(n, "stats") # noqa: E731 + assert vd.rows(est(600), [est(1000), est(1000)]) == [], "估算值跌了一次：先不报" + assert [f.kind for f in vd.rows(est(600), [est(650), est(1000)])] == ["rows_drop"], "连续两次都跌：报" + assert vd.rows(_rows(600), [_rows(1000, "stats")]) == [], "数出来的和估算的口径不同：不比" + assert vd.rows(vd.Obs("rows", "count", {"rows": None, "at_least": 1001}, AT), [_rows(5000)]) == [], \ + "只知道「至少多少行」的不比"