diff --git a/backend/app/api/catalog.py b/backend/app/api/catalog.py index 0311bc4c..46908acb 100644 --- a/backend/app/api/catalog.py +++ b/backend/app/api/catalog.py @@ -292,7 +292,10 @@ async def profile_catalog_tables(source_id: str, payload: CatalogProfileIn, raise HTTPException(409, str(e)) from e rows = [_profile_table_out(t) for t in report.tables] note = None - if not rows: + if not rows and report.empty_tables: + note = (f"有待核对关联关系的 {len(report.empty_tables)} 张表都是空表,没有可以核实的数据。" + "请在数据有了之后再剖析,或指定要剖析的表") + elif not rows: note = "目录里还没有待核实的关联关系,请先起草数据目录,或指定要剖析的表" return { "profiled_at": report.profiled_at, @@ -302,6 +305,8 @@ async def profile_catalog_tables(source_id: str, payload: CatalogProfileIn, "stopped": report.stopped, "tables": rows, "total": {k: sum(r[k] for r in rows) for k in ("added", "updated", "removed")}, + # 不指定表时挑表途中确知是空表、跳过了的表(指定了表时总是空列表) + "empty_tables": report.empty_tables, "note": note, } diff --git a/backend/app/data/catalog_profile.py b/backend/app/data/catalog_profile.py index fad61cbe..82cb980e 100644 --- a/backend/app/data/catalog_profile.py +++ b/backend/app/data/catalog_profile.py @@ -440,8 +440,10 @@ async def estimate_size(run: _Runner, dialect: SqlDialect, meta: dict[str, Any], # 一次剖析:上下文、结果 # ========================================================================== -#: 不指定表时剖析几张:按使用次数取目录里有待核实关系的表 +#: 不指定表时剖析几张:目录里有要核对的关系的表,还有待核实(推断)关系的在前,再按使用次数;确知是空表的跳过 PROFILE_DEFAULT_TABLES = 10 +#: 不指定表时,为跳过空表最多先估几张候选表的行数;估行数的查询最多用掉查询次数上限的一半,免得都花在挑表上 +PROFILE_PROBE_TABLES = 30 #: IN 列表每批最多几个值:Oracle 的 IN 列表不能超过 1000 项(ORA-01795),各家统一按它分批 IN_CHUNK = 1000 #: 剖析写的备注都以它开头(「数据剖析:」):人工确认过的关系,只覆盖剖析自己写的备注,不碰人写的(认的时候连 @@ -512,6 +514,8 @@ class ProfileReport: tables: list[TableProfile] = field(default_factory=list) queries_used: int = 0 stopped: str | None = None + #: 不指定表时,挑表途中估过行数、确知是空表而跳过的表(表结构顺序)。估行数的查询算在 queries_used 里 + empty_tables: list[str] = field(default_factory=list) @dataclass @@ -1156,7 +1160,44 @@ async def _pick_tables(session: AsyncSession, source: Any, tables: list[str] | N candidates = [name for name, meta in all_tables.items() if not (meta or {}).get("is_view") and name in entries and _relation_candidates(entries[name].notes)] - return sorted(candidates, key=lambda n: (-usage.get(n, 0), order[n]))[:PROFILE_DEFAULT_TABLES] + # 还有待核实(推断)关系的表排在前面:剖析就是来核实它们的;关系都核实过、人工确认过的再剖析只是补覆盖率。 + # 同样待核实的按使用次数,再按表结构顺序。返回全部候选,空表由 _skip_empty 估过行数再跳过 + return sorted(candidates, key=lambda n: (not _pending(entries[n].notes), -usage.get(n, 0), order[n])) + + +def _pending(notes: dict[str, Any] | None) -> bool: + """这张表还有没有待核实的关系:要核对的关系里有还是推断(proposed)的。""" + return any(r.get("status") == "proposed" for r in _relation_candidates(notes)) + + +async def _skip_empty(cx: _Context, candidates: list[str], report: ProfileReport) -> list[TableProfile]: + """不指定表时:按顺序估候选表的行数,确知是空表的跳过(记进 report.empty_tables),凑够 PROFILE_DEFAULT_TABLES + 张为止。以前不估,所有表使用次数都是 0 时按表结构顺序取前 10 张,结果大半是空表、一条关系都核实不了。 + + 估行数的查询算进这张表的查询数,行数留在 cx.sizes 里,剖析时不再查。最多估 PROFILE_PROBE_TABLES 张、最多用掉 + 查询次数上限的一半;超过了、或者整次剖析已经停下,剩下的候选不估、照顺序排上(真是空表的,剖析时各项检查会 + 记下「空表,无法剖析」)。 + """ + picked: list[TableProfile] = [] + probed = 0 + for name in candidates: + if len(picked) >= PROFILE_DEFAULT_TABLES: + break + result = TableProfile(table=name) + if (cx.run.stopped is None and probed < PROFILE_PROBE_TABLES + and cx.run.used < cx.settings.max_queries // 2): + probed += 1 + before = cx.run.used + try: + size = await cx.size(name, result) + except _Stop: + size = None + result.queries = cx.run.used - before + if size is not None and size.rows == 0: + report.empty_tables.append(name) + continue + picked.append(result) + return picked async def profile_catalog(session: AsyncSession, row: Any, source: Any, *, tables: list[str] | None = None, @@ -1200,8 +1241,10 @@ async def _profile(session: AsyncSession, row: Any, source: Any, tables: list[st tables=(getattr(source, "schema_cache", None) or {}).get("tables") or {}, masked={c.lower() for c in data_engine.masked_columns(getattr(row, "options", None))}) entries = await catalog.read_catalog(session, source.id) - for table in await _pick_tables(session, source, tables, entries, report): - result = TableProfile(table=table) + picked = await _pick_tables(session, source, tables, entries, report) + results = [TableProfile(table=t) for t in picked] if tables else await _skip_empty(cx, picked, report) + for result in results: + table = result.table report.tables.append(result) if run.stopped: detail = _STOP_DETAIL[run.stopped].format(max_queries=settings.max_queries, @@ -1211,7 +1254,7 @@ async def _profile(session: AsyncSession, row: Any, source: Any, tables: list[st before = run.used entry = entries.get(table) out = await _profile_table(cx, table, entry.notes if entry else {}, result) - result.queries = run.used - before + result.queries += run.used - before await _write(session, source.id, table, out, result, at=at, actor=actor) report.queries_used = run.used report.stopped = run.stopped @@ -1221,6 +1264,6 @@ async def _profile(session: AsyncSession, row: Any, source: Any, tables: list[st __all__ = ["CODES_MAX", "IN_CHUNK", "MAX_CODE_LEN", "MAX_LITERAL_LEN", "NOTE_PREFIX", "PROFILE_DEFAULT_TABLES", - "PROFILE_OPTION", "ProfileBusy", "ProfileDisabled", "ProfileReport", "ProfileSettings", "ProfileSkip", - "SqlDialect", "TableProfile", "TableSize", "ensure_enabled", "estimate_size", "profile_catalog", - "profile_settings", "profile_settings_problem", "sql_literal"] + "PROFILE_OPTION", "PROFILE_PROBE_TABLES", "ProfileBusy", "ProfileDisabled", "ProfileReport", + "ProfileSettings", "ProfileSkip", "SqlDialect", "TableProfile", "TableSize", "ensure_enabled", + "estimate_size", "profile_catalog", "profile_settings", "profile_settings_problem", "sql_literal"] diff --git a/backend/tests/test_catalog_profile.py b/backend/tests/test_catalog_profile.py index fdc605bc..de870387 100644 --- a/backend/tests/test_catalog_profile.py +++ b/backend/tests/test_catalog_profile.py @@ -431,6 +431,69 @@ async def test_default_tables_are_used_tables_with_relation_candidates(client, m assert "parks" not in names and "channels" not in names # 没有待核实关系的表不选 +async def _candidates(sid: str) -> tuple[list[str], set[str]]: + """默认挑表的候选(目录里有要核对的关系的表,按表结构顺序)和其中还有待核实(推断)关系的那些。""" + async with SessionLocal() as session: + row = await session.get(DataSource, sid) + entries = await catalog.read_catalog(session, sid) + order = list((row.schema_cache or {}).get("tables") or {}) + rels = {name: catalog_profile._relation_candidates(entries[name].notes) for name in order if name in entries} + names = [name for name in order if rels.get(name)] + return names, {name for name in names if any(r["status"] == "proposed" for r in rels[name])} + + +async def test_default_pick_skips_empty_tables_and_prefers_pending_relations(client, make_source, tmp_path, + monkeypatch): + """B1:表都没被运行查询过(使用次数都是 0)时,以前按表结构顺序取前几张,结果大半是空表(景区库里排在前面的 + 候选表正好多数没有数据)。现在跳过行数估计为 0 的表(记在 empty_tables 里),并且先挑还有待核实关系的表, + 不把刚核实过的再剖析一遍。默认张数调成 3,好在一个小库上看出先后。""" + monkeypatch.setattr(catalog_profile, "PROFILE_DEFAULT_TABLES", 3) + path = scenic.build(tmp_path / "b1.db") + db = sqlite3.connect(path) + try: + tables = [r[0] for r in db.execute("SELECT name FROM sqlite_master WHERE type = 'table' ORDER BY rowid")] + empty = {t for t in tables if db.execute(f'SELECT COUNT(*) FROM "{t}"').fetchone()[0] == 0} + finally: + db.close() + sid = await make_source(str(path), options=_on(), draft=tables) + names, pending = await _candidates(sid) + full = [name for name in names if name not in empty] + # 前提:按表结构顺序的前 3 张候选里有空表(以前就会挑中它们),有数据、还待核实的候选不止 6 张 + assert set(names[:3]) & empty + assert len([name for name in full if name in pending]) >= 6 + + first = (await client.post(_url(sid), json={}, headers=ACTOR)).json() + picked = [t["table_name"] for t in first["tables"]] + assert len(picked) == 3 and not set(picked) & empty, picked + assert first["empty_tables"] and set(first["empty_tables"]) <= empty + assert all(t["queries"] > 0 for t in first["tables"]) + assert first["queries_used"] >= sum(t["queries"] for t in first["tables"]) + len(first["empty_tables"]) + + # 第二次:先挑还有待核实关系的表;第一次核实过的(关系都升为有确证了)排到后面,这次挑不到 + names, pending = await _candidates(sid) + assert not set(picked) & pending # 第一次挑的三张,关系都核实了 + second = (await client.post(_url(sid), json={}, headers=ACTOR)).json() + again = [t["table_name"] for t in second["tables"]] + assert len(again) == 3 and all(name in pending for name in again), (again, pending) + assert not set(again) & set(picked) and not set(again) & empty + + +async def test_profile_note_when_every_candidate_is_empty(client, make_source, tmp_path): + path = scenic.build(tmp_path / "b1-empty.db") + sid = await make_source(str(path), options=_on(), draft=[]) + names, _ = await _candidates(sid) + db = sqlite3.connect(path) + try: + for name in names: + db.execute(f'DELETE FROM "{name}"') + db.commit() + finally: + db.close() + body = (await client.post(_url(sid), json={}, headers=ACTOR)).json() + assert body["tables"] == [] and set(body["empty_tables"]) == set(names) + assert "空表" in body["note"] + + async def test_unknown_table_and_missing_source(client, make_source, scenic_db): sid = await make_source(scenic_db, options=_on(), draft=["visits"]) r = await client.post(_url(sid), json={"tables": ["nope"]}, headers=ACTOR) diff --git a/frontend/src/types.ts b/frontend/src/types.ts index f2e4f915..4729e179 100644 --- a/frontend/src/types.ts +++ b/frontend/src/types.ts @@ -2943,6 +2943,11 @@ export interface CatalogProfileOut { stopped: CatalogProfileStop | null tables: CatalogProfileTable[] total: { added: number; updated: number; removed: number } + /** + * 不指定表时,挑表途中估过行数、确知是空表而跳过的表(老后端没有这个键;指定了表时是空列表)。 + * 估行数的查询算在 queries_used 里,不算在各表的 queries 里 + */ + empty_tables?: string[] /** 一张表都没有剖析时的说明 */ note: string | null }