W3-a 关联覆盖率下降、清理目录跨表、扩容器集成测试到对象索引和取值观察

- 剖析核对关联关系时，覆盖率从够有确证的线（95%）掉到线以下的记一条 coverage_drop（取值变化，warning）：剖析本身照旧
  按规则降回推断（人工确认的只补上新的覆盖率），这里只把「覆盖率下降」留进变化记录。人点的剖析也记（变化记录全站一份）。
  关系的发现里多带 previous_coverage。
- 处理动作「清理目录」逐表清：这张表上说到被删的列的项，加上别的表指向它的关联关系（表删了时尤其如此）；记这张表的
  目录版本，别的表也动了的另记 catalog_versions。
- 剖析任务的逐表进度多一类 observe（取值观察的四类合成一组），前端「取值观察 N/M」用它。
- 定时维护补一条测试：AGENTLAB_SCHEDULER 关着时调度循环不起，到点的维护不触发、不建任务。
- scripts/dialect-it.sh / dialect_it.py 扩到第 2 波的对象索引字典查询（走字典查询不退回 Inspector、类型、表注释、行数
  估算、Oracle 小写名的 quote）和这一波的取值观察（经剖析引擎只观察：行数、小表码值整表、大表 PG / Oracle 块采样、MySQL
  前 N 行、最新业务日期排除无效日期，max_date_sql 直接跑）。夹具的订单表加了表注释。本机跑了一次四种库的容器
  （postgres:16、mysql:8.4、mariadb:11、gvenzl/oracle-free:23-slim）：121 项全过、0 项不通过，跑完容器已删、镜像留着。

验证：后端全量 14046 通过（11 跳过、8 预期失败）。

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}
 
 

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，否则

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"]

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

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)]

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)

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 起一次要一两分钟。要用户同意才在本机跑。跑完删容器、留镜像。
 #

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)