diff --git a/README.md b/README.md index f3976735..9bcc471e 100644 --- a/README.md +++ b/README.md @@ -928,7 +928,7 @@ be brought back by the next drafting or inference. | Review queue | Lists only inferred items, ordered by "how many times runs used the table × how much the item affects numbers × source": valid-record conditions, dedup rules, business dates, and relationships come first, business keys and measure types next, and Chinese names, meanings, and units last, with each item stating why it's ranked there; filter by category and table, and multi-select to confirm, reject, or undo review in bulk (optimistic locking per table, with conflicts flagged one by one); you can see which runs used a table, and how many items have been confirmed for frequently used tables, for all tables, and for the items that affect numbers | | Table Copilot (「表助手」) | Chat with Copilot on the table details page to refine this one table: it can see the schema, the current state of the catalog (each item's source and status), relationships, profiling results, drafting instructions, and the columns left empty in the last drafting (masked columns: name only); it proposes changes in batches following your instructions, each with before, after, and basis (validated like the catalog: formulas aren't accepted, and column names must exist in the schema), and asks you back when it isn't sure; only on sources with profiling on does it run queries through the read-only guard to verify (at most 5 per request, with results masked first). Proposals accumulate in Pending changes (「待保存的修改」) and are saved in one go after confirmation (at most 200 items at a time), recorded as confirmed by a person, with each item's basis saved separately as that item's Basis at confirmation (「确认时的依据」) (kept apart from notes and visible in the table details; removed when undoing review returns the item to inferred). Conversations aren't stored on the server, and it changes only the catalog, never the business database | | Export and import | Exports one JSON file (the catalog of all tables, each item's status, and drafting instructions) to import into another instance or data source: first preview which tables and columns match, and whether each item will be added, updated, or kept, with conflicts and skips listed one by one; items a person confirmed or rejected are carried over as they are, where the target already has a confirmed item with a different value, the target's is kept and recorded as a conflict, inferred items in the file only fill gaps, and columns not in the schema are skipped with the reason stated; import isn't possible while drafting or profiling is in progress. Drafting instructions are written only when the target has none | -| Data profiling | Off by default, enabled per data source. Runs as a background job, table by table: for each table you can see check progress, query count, elapsed time, and which item is being checked; closing the dialog doesn't stop it, and you can stop it or retry only the unfinished tables; on service restart, unfinished jobs are marked interrupted and don't resume by themselves. The budget is per table (by default at most 40 queries and 60 seconds per table), and when one table uses up its budget only that table stops, while the next one gets a full budget; there's no cap for the job as a whole, which relies on stop-loss rules and manual stopping: it stops after 3 consecutive failed queries, or when 7 of the last 10 queries time out, so as not to put load on the business database. Queries are sent one at a time, read-only, through the guard; large tables use statistics to estimate row counts, and masked columns aren't sampled; every query sent leaves an audit trail (with values replaced by "?", and the sampled key values never enter AgentLab's database). Sampling verifies inferred relationships, filling in coverage and cardinality, and the ones that hold up are promoted to verified; status columns get code-value candidates, with meanings left for a person to fill in. By default it profiles "tables that need profiling" (「需要剖析的表」): those never profiled, not finished last time, or that have since gained relationships awaiting verification or new code-value or date columns; on the same data source, drafting and profiling queue up serially, and the scope is set from the catalog as it stands when profiling starts, so you can simply click profile right after drafting finishes. The table list has a Profiling (「剖析」) column, and the table details show the last profiling result | +| Data profiling | Off by default, enabled per data source. Runs as a background job, table by table: for each table you can see check progress, query count, elapsed time, and which item is being checked; closing the dialog doesn't stop it, and you can stop it or retry only the unfinished tables; on service restart, unfinished jobs are marked interrupted and don't resume by themselves. The budget is per table (by default at most 40 queries and 60 seconds per table), and when one table uses up its budget only that table stops, while the next one gets a full budget; there's no cap for the job as a whole, which relies on stop-loss rules and manual stopping: it stops after 3 consecutive failed queries, or when 7 of the last 10 queries time out, so as not to put load on the business database. By default it profiles one table at a time; in the data source settings you can raise this to 2–3 tables at once (so at most that many queries hit the business database at any moment). Within each table, queries are sent one at a time, read-only, through the guard; large tables use statistics to estimate row counts, and masked columns aren't sampled; every query sent leaves an audit trail (with values replaced by "?", and the sampled key values never enter AgentLab's database). Sampling verifies inferred relationships, filling in coverage and cardinality, and the ones that hold up are promoted to verified; status columns get code-value candidates, with meanings left for a person to fill in. By default it profiles "tables that need profiling" (「需要剖析的表」): those never profiled, not finished last time, or that have since gained relationships awaiting verification or new code-value or date columns; on the same data source, drafting and profiling queue up serially, and the scope is set from the catalog as it stands when profiling starts, so you can simply click profile right after drafting finishes. The table list has a Profiling (「剖析」) column, and the table details show the last profiling result | | Corrections in conversation | When a user says "status=9 means voided" (「status=9 表示作废」), Copilot produces a catalog change proposal listing before, after, and the reason; it's saved only after a person confirms, and the reason is recorded as the item's Basis at confirmation (「确认时的依据」); formulas aren't accepted and are left to caliber cards | | Versions and reminders | The catalog is frozen into the evidence along with the schema snapshot, so you can look up which version every run used. Publishing records the catalog versions of the tables a template references, and a formal run gives a reminder when the catalog has changed; where an agent writes SQL only at run time, there's a separate reminder for "possibly involved" (「可能涉及」). It only reminds and never blocks the run | diff --git a/README.zh-CN.md b/README.zh-CN.md index 2548eab9..e2309f00 100644 --- a/README.zh-CN.md +++ b/README.zh-CN.md @@ -644,7 +644,7 @@ curl -H "Authorization: Bearer agl_…" -H "Idempotency-Key: order-20261016-001" | 审阅队列 | 只列推断的项，按「表被运行用到的次数 × 这一项多影响数字 × 来源」排序：有效记录条件、去重规则、业务日期、关联关系最前，业务主键、度量类型次之，中文名、含义、单位靠后，每项写明排在这里的理由；按类别和表筛选，多选批量确认、驳回或撤销审阅（逐表乐观锁，冲突逐条标出）；看得到一张表被哪些运行用过，以及常用表、全部和影响数字的项各确认了多少 | | 表助手 | 表详情页里和助手对话完善这一张表：它看得到表结构、目录现状（每项的来源和状态）、关联关系、剖析结果、起草要求和上次起草留空的列（遮罩列只给列名），按你的指示成批提出修改，每项带改前、改后和依据（照目录的校验：公式不收、列名要在表结构里），拿不准的会反过来问你；只在开了剖析的源上经只读守卫跑查询核实（每次请求最多 5 条，结果先遮）。提案攒在「待保存的修改」里，确认后一次保存（一次最多 200 项），记为人工确认，每项的依据另存为这一项「确认时的依据」（和备注分开，表详情里看得到；撤销审阅回到推断时去掉）。对话不存服务端，只改目录、不碰业务库 | | 导出与导入 | 导出成一个 JSON 文件（全部表的目录、各项状态和起草要求），在另一个实例或数据源上导入：先预览哪些表、列对得上，每一项会新增、更新还是保留，冲突和跳过的逐条列出；人确认、驳回过的项照带，目标里已经确认过而值不同的保留目标、记成冲突，文件里推断的项只填空缺，表结构里没有的列跳过并写原因；有起草或剖析在进行时不能导入。起草要求只在目标没有时写入 | -| 数据剖析 | 默认关闭，按数据源开启。作为后台任务逐表进行：每张表看得到检查项进度、查询数、用时和正在核对哪一项，关掉对话框不停，可以停止、只重试没做完的表；服务重启时没跑完的任务标成中断，不会自己续跑。预算按表算（缺省每张表最多 40 条查询、60 秒），一张表用完只停它自己，下一张照样满额；整次不设上限，靠止损和手动停止：连续 3 条查询失败，或最近 10 条有 7 条超时就停，不给业务库加压。查询一条一条发、只读、经过守卫，大表用统计信息估算行数，遮罩的列不取样；发过的每条查询都留痕（取值换成「?」，抽到的键值不进 AgentLab 的库）。抽样核实推断出的关联，补上覆盖率和基数，站得住的升为已验证；状态列给出码值候选，含义等人填写。缺省剖析「需要剖析的表」：从没剖析过、上次没做完，或之后又出现了待核实关联、新的码值或日期列的；同一个数据源上起草和剖析排队串行，范围在开始剖析时按当时的目录定，所以起草完接着点剖析就行。表清单有「剖析」一列，表详情有上次剖析的结果 | +| 数据剖析 | 默认关闭，按数据源开启。作为后台任务逐表进行：每张表看得到检查项进度、查询数、用时和正在核对哪一项，关掉对话框不停，可以停止、只重试没做完的表；服务重启时没跑完的任务标成中断，不会自己续跑。预算按表算（缺省每张表最多 40 条查询、60 秒），一张表用完只停它自己，下一张照样满额；整次不设上限，靠止损和手动停止：连续 3 条查询失败，或最近 10 条有 7 条超时就停，不给业务库加压。缺省一张表做完再做下一张，数据源设置里可以改成同时剖析 2–3 张（同一时刻对业务库最多这么多条查询）；每张表的查询一条一条发、只读、经过守卫，大表用统计信息估算行数，遮罩的列不取样；发过的每条查询都留痕（取值换成「?」，抽到的键值不进 AgentLab 的库）。抽样核实推断出的关联，补上覆盖率和基数，站得住的升为已验证；状态列给出码值候选，含义等人填写。缺省剖析「需要剖析的表」：从没剖析过、上次没做完，或之后又出现了待核实关联、新的码值或日期列的；同一个数据源上起草和剖析排队串行，范围在开始剖析时按当时的目录定，所以起草完接着点剖析就行。表清单有「剖析」一列，表详情有上次剖析的结果 | | 对话里纠正 | 用户说「status=9 表示作废」，助手给出一条目录修改提案，列出改前、改后和理由，人确认后才保存，理由记为这一项的「确认时的依据」；公式不收，留给口径卡 | | 版本与提醒 | 目录随表结构快照冻结进证据，每次运行用的是哪一版查得到。发布时记下模板引用的表的目录版本，正式运行时目录有变化就提醒；agent 运行时才写 SQL 的，按「可能涉及」单独提醒。只提醒，不拦运行 | diff --git a/backend/app/api/catalog.py b/backend/app/api/catalog.py index 250a48ad..34c4339c 100644 --- a/backend/app/api/catalog.py +++ b/backend/app/api/catalog.py @@ -499,7 +499,7 @@ async def _start_profile(session: AsyncSession, row: DataSource, source: Any, *, async def create_profile_job(source_id: str, payload: ProfileJobIn, x_actor: str | None = Header(default=None), session: AsyncSession = Depends(get_session)) -> dict[str, Any]: """建后台剖析任务，马上返回（202，任务是排队中）。tables 非空时范围是 selected（表结构里没有、视图、没有可剖析的项照样 - 建进任务、状态 skipped），否则按 scope（缺省 needed）挑。同源有排队或进行中的剖析任务 409 profile_busy（说清进行到第几张； + 建进任务、状态 skipped），否则按 scope（缺省 needed）挑。同源有排队或进行中的剖析任务 409 profile_busy（说清做完了几张； 界面改去 GET /profile-jobs/active 显示那一个），按范围挑出 0 张表 409 profile_nothing；同源有起草任务时照建，排在它后面 （waiting_for = "draft"），起草结束后自动开始。""" if payload.tables and len(payload.tables) > catalog_profile_jobs.MAX_TABLES: diff --git a/backend/app/data/catalog_jobs.py b/backend/app/data/catalog_jobs.py index 9a0826b3..4f2e7c24 100644 --- a/backend/app/data/catalog_jobs.py +++ b/backend/app/data/catalog_jobs.py @@ -370,7 +370,7 @@ async def ensure_draft_free(session: AsyncSession, row: Any, *, then: str = THEN async def ensure_idle(session: AsyncSession, row: Any, kind: str = "import") -> None: """同步写操作（kind：import 导入目录、sync_draft 同步起草）要求源空闲：没有在起草（DraftBusy，同 ensure_draft_free）， - 也没有排队或进行中的剖析任务（catalog_profile_jobs.ProfileBusy，说法里写明进行到第几张）。说法最后一句按 kind 说 + 也没有排队或进行中的剖析任务（catalog_profile_jobs.ProfileBusy，说法里写明做完了几张）。说法最后一句按 kind 说 这次要做的事（_then）。""" from app.data import catalog_profile_jobs diff --git a/backend/app/data/catalog_profile.py b/backend/app/data/catalog_profile.py index 7ea86279..171b8f0a 100644 --- a/backend/app/data/catalog_profile.py +++ b/backend/app/data/catalog_profile.py @@ -16,7 +16,8 @@ hooks 回调（后台任务据此把进度落库），整张表结束把结论 - 默认关闭，按数据源在 options.catalog_profile 里开启（profile_settings）； - 每条查询先过守卫的只读判定（不论数据源本身可写与否），再交给查询层 run_query：同一套连接、守卫和 - 时限，不另开连接；查询一条一条发，不并发； + 时限，不另开连接；一张表的查询一条一条发，不并发；同时剖析几张表按设置（table_concurrency，缺省 1、最多 + MAX_TABLE_CONCURRENCY），所以同一时刻对业务库最多这么多条查询； - 单条查询的时限取设置和数据源查询时限中小的，且不超过这张表剩下的时长； - 每张表有自己的预算（TableBudget：查询次数、时长），用完只停这张表，剩下的项记「这张表的查询次数（时长）用完」，下一张 表拿满额。整个任务一个查询器（SourceRunner）管任务级止损：连续 3 条查询失败（多半是连接断了、没有权限）、最近 10 条里 @@ -87,6 +88,9 @@ class ProfileSettings: #: 行数不超过它的表才做 COUNT(DISTINCT)、MIN/MAX 这类整表统计；更大的、或者不知道多大的表只看前若干行。 #: 填 0 表示一律不做整表统计 max_scan_rows: int = 100_000 + #: 同时剖析几张表（每张表的查询仍一条一条发）。缺省 1：一张做完再做下一张；大于 1 时业务库同时要应付这么多条查询， + #: 和这个源上的分析共用连接池，所以上限是 MAX_TABLE_CONCURRENCY + table_concurrency: int = 1 def to_dict(self) -> dict[str, Any]: return asdict(self) @@ -105,7 +109,11 @@ _NUMBER_FIELDS: dict[str, tuple[str, float, float, bool]] = { "query_timeout_s": ("单条查询时限（秒）", 1, 60, False), "sample_size": ("核对键值数", 10, 10_000, True), "max_scan_rows": ("整表统计行数上限", 0, 10_000_000, True), + "table_concurrency": ("同时剖析的表数", 1, 3, True), } +#: 同时剖析的表数的上限：数据源的连接池常驻 3 条、临时再借 2 条（data/engine.EngineCache），和这个源上的分析共用， +#: 剖析最多占常驻的 3 条，临时的留给分析 +MAX_TABLE_CONCURRENCY = int(_NUMBER_FIELDS["table_concurrency"][2]) _ENABLED_LABEL = "开启数据剖析" _KNOWN_KEYS = frozenset({"enabled", *_NUMBER_FIELDS}) #: 以前的整次预算（一次剖析最多几条查询、总时长）。现在预算按表算，这两项不再有意义：启动时从库里删掉 @@ -979,7 +987,8 @@ class _TableOutcome: @dataclass class ProfileContext: """一个剖析任务共用的：源（上传源是绑定当前快照的视图）、设置、方言、查询器（任务级止损在它身上）、表结构、遮罩的 - 列（小写，任务每张表开始前按数据源最新的设置换掉）、已经估过的行数（父表的行数核对关系时会用到，不重复估）。""" + 列（小写，任务每张表开始前按数据源最新的设置换掉）、已经估过的行数（父表的行数核对关系时会用到，不重复估）。 + 同时剖析的几张表共用这一份（都在同一个事件循环里）。""" source: Any settings: ProfileSettings @@ -989,6 +998,8 @@ class ProfileContext: #: 遮罩的列（小写）。遮罩按列名生效、不分大小写、不分表，和证据面板的比法一致 masked: set[str] sizes: dict[str, TableSize] = field(default_factory=dict) + #: 正在估哪张表的行数（见 _TableCx.size）：同时剖析的几张表要估同一张时只估一次 + size_locks: dict[str, asyncio.Lock] = field(default_factory=dict) #: 对象索引里探查范围外的对象（name_key）：关系指向它们时说「在探查范围外」，不说「表结构里没有」 outside: frozenset[str] = frozenset() @@ -1080,15 +1091,19 @@ class _TableCx: await self.hooks.query(self.result) async def size(self, table: str, result: TableResult) -> TableSize: - """表有多大，一个任务里每张表只估一次。途中没查成的记进 result（行数估算这一项）。""" + """表有多大，一个任务里每张表只估一次。途中没查成的记进 result（行数估算这一项）。同时剖析的几张表要估同一张 + （同一张父表，或者它自己正在剖析）时后来的等先来的估完、用它的结果，查询记在先来的那张表上；先来的没估成（停止、 + 预算用完）就自己估。""" if table not in self.job.sizes: - async def run(sql: str, **kw: Any) -> QueryResult: - return await self.run(sql, kind="row_estimate", target=table, **kw) + async with self.job.size_locks.setdefault(table, asyncio.Lock()): + if table not in self.job.sizes: + async def run(sql: str, **kw: Any) -> QueryResult: + return await self.run(sql, kind="row_estimate", target=table, **kw) - size, skips = await estimate_size(run, self.dialect, self.tables[table], table, - max_scan_rows=self.settings.max_scan_rows) - self.job.sizes[table] = size - result.skipped += [ProfileSkip("row_estimate", table, s.reason, s.detail) for s in skips] + size, skips = await estimate_size(run, self.dialect, self.tables[table], table, + max_scan_rows=self.settings.max_scan_rows) + self.job.sizes[table] = size + result.skipped += [ProfileSkip("row_estimate", table, s.reason, s.detail) for s in skips] return self.job.sizes[table] async def size_with_data(self, table: str, result: TableResult) -> TableSize: diff --git a/backend/app/data/catalog_profile_jobs.py b/backend/app/data/catalog_profile_jobs.py index 6b25d9a7..fe19a8d7 100644 --- a/backend/app/data/catalog_profile_jobs.py +++ b/backend/app/data/catalog_profile_jobs.py @@ -8,17 +8,19 @@ data/catalog_job_runtime.py），逐表剖析（引擎在 data/catalog_profile.p 任务、状态 skipped）、需要剖析的表（needed：从没剖析过、上次剖析没做完、剖析之后冒出了新的待核实关系，见 due_reason）、 全部表（all）。后两种只挑表结构里、不是视图、有可剖析的项的表，有待核实关系的在前，再按使用次数、表结构顺序；挑出 0 张 抛 ProfileNothing。剖析设置随任务记下快照，每张表的检查项按当时的目录先列一遍。同一个源同时只有一个排队或进行中的 - 剖析任务（ProfileBusy，说法里写明进行到第几张；部分唯一索引兜底）；和同源的起草任务按建立的先后排队（每个源一条队）， + 剖析任务（ProfileBusy，说法里写明做完了几张；部分唯一索引兜底）；和同源的起草任务按建立的先后排队（每个源一条队）， 起草在前时 waiting_for 是 "draft"。建任务时顺手清掉超出保留数的旧任务（prune）。 -- **在后台剖析**（manager.start）：轮到了才开始。每张表开始前重读数据源——停用了、关掉了剖析就停下，剩下的表 cancelled、 - error 写明；遮罩按最新的设置——再逐表调 catalog_profile.profile_table（它读的是那时目录里的最新内容：起草刚提的关系也 - 核对）。检查项、正在做哪一项、查询数、发过的查询随时落库；整张表的结果和写目录在同一个事务里，停止时也不会被打断。 -- **任务级止损**（替代以前的整次总量上限）：连续 3 条查询失败、最近 10 条里 7 条超时，整个任务停下（stop_reason、error）， - 正在做的那张表写入已做完的项，后面的表 cancelled、error 带同一句原因；最近 6 条里 4 条超时时 notice 有一句提醒，好转后 - 清掉。任务最后的状态按逐表算：有 failed、partial、cancelled 的表是 partial，否则 succeeded；任务整体出错（数据源没了、 - 快照被改动、意外异常）是 failed。 -- **停止**（cancel）：排队的直接 cancelled；进行中的停掉正在等的那条查询（库那边按单条时限截止），那张表做完的项照常写进 - 目录、没做的保留旧结论，后面的表 cancelled，任务 cancelled。 +- **在后台剖析**（manager.start）：轮到了才开始。按任务里的顺序逐表调 catalog_profile.profile_table（它读的是那时目录里 + 的最新内容：起草刚提的关系也核对），同时最多剖析设置里 table_concurrency 张（缺省 1，一张做完再做下一张；每张表的查询 + 仍一条一条发），一张结束就开始下一张。每张表开始前重读数据源——停用了、关掉了剖析就不再开始新的表，正在做的做完， + 剩下的表 cancelled、error 写明；遮罩按最新的设置。检查项、正在做哪一项、查询数、发过的查询随时落在各自那张表上；整张表 + 的结果和写目录在同一个事务里，停止时也不会被打断。 +- **任务级止损**（替代以前的整次总量上限）：连续 3 条查询失败、最近 10 条里 7 条超时（同时剖析的几张表的查询合起来算）， + 整个任务停下（stop_reason、error），正在做的几张表写入已做完的项，后面的表 cancelled、error 带同一句原因；最近 6 条里 + 4 条超时时 notice 有一句提醒，好转后清掉。任务最后的状态按逐表算：有 failed、partial、cancelled 的表是 partial，否则 + succeeded；任务整体出错（数据源没了、快照被改动、意外异常）是 failed，同时在做的另几张表照停止收尾、原因写这一句。 +- **停止**（cancel）：排队的直接 cancelled；进行中的停掉正在等的查询（库那边按单条时限截止），正在做的几张表做完的项照常 + 写进目录、没做的保留旧结论，后面的表 cancelled，任务 cancelled。 - **重试**（retry_tables + create_job）：新任务，只含原任务里 failed、partial、cancelled 的表（按原来的顺序），范围 selected， retry_of 记原任务；设置取数据源当前的。 - **服务重启**（startup）：排队、进行中的标 interrupted，没结束的表 cancelled、error「服务重启，剖析中断」（进程没了时正在 @@ -44,6 +46,8 @@ data/catalog_job_runtime.py），逐表剖析（引擎在 data/catalog_profile.p """ from __future__ import annotations +import asyncio +import collections import logging import time from typing import Any @@ -94,7 +98,7 @@ def unsaved_error(why: str, done: int, rest: int) -> str: class ProfileBusy(Exception): - """这个数据源已经有排队或进行中的剖析任务。message 给人看（写明进行到第几张）。""" + """这个数据源已经有排队或进行中的剖析任务。message 给人看（写明做完了几张）。""" class ProfileFinished(Exception): @@ -501,15 +505,15 @@ THEN_PROFILE = "再开始新的剖析" def busy_message(row: Any, brief: dict[str, Any] | None, *, table: str | None = None, position: int | None = None, then: str = THEN_PROFILE) -> str: - """同源已有剖析任务时的说法：写明进行到第几张；从表详情点「剖析这张表」时再说这张表在不在那个任务里、排第几；最后 - 一句按调用方要做的事说（then）。""" + """同源已有剖析任务时的说法：写明做完了几张（同时剖析几张表时「第几张」说不清，只说做完几张）；从表详情点「剖析这张 + 表」时再说这张表在不在那个任务里、排第几；最后一句按调用方要做的事说（then）。""" name = getattr(row, "name", "") or "" if brief is None: text = f"数据源「{name}」正在剖析" elif brief["status"] == "queued": text = f"数据源「{name}」已有一个排队中的剖析任务（共 {brief['total']} 张表）" else: - text = f"数据源「{name}」正在剖析：第 {min(brief['done'] + 1, brief['total'])} / {brief['total']} 张表" + text = f"数据源「{name}」正在剖析：已完成 {brief['done']} / {brief['total']} 张表" if table is not None: text += f"，{table} 在这次剖析里，排第 {position} 张" if position else f"，{table} 不在这次剖析里" return text + f"。请等这次剖析结束（或先停止它）{then}" @@ -880,6 +884,36 @@ async def startup(*, live: Any = None) -> int: # ========================================================================== +def _job_error(e: BaseException) -> str: + """任务整体意外出错时的整句（任务、没结束的表的 error）。""" + from app.core.errors import describe_exception, first_line + + return JOB_ERROR.format(line=first_line(e) or describe_exception(e)) + + +def _elapsed(started: float) -> int: + return int((time.monotonic() - started) * 1000) + + +async def _drain(tasks: Any) -> None: + """停掉这几个协程，等它们都结束（正在剖析的表各自收尾、把做完的项写进目录）。等的时候自己又被停止也等完，再把停止 + 往上交：不然 JobRun 收尾时这几张表还没写完，会被当成没写进目录。""" + tasks = list(tasks) + for task in tasks: + task.cancel() + stopped = False + while pending := [t for t in tasks if not t.done()]: + try: + await asyncio.wait(pending) + except asyncio.CancelledError: + stopped = True + for task in tasks: + if not task.cancelled(): + task.exception() # 取走，免得「异常没人取」的告警；要往上交的那个调用方已经拿着 + if stopped: + raise asyncio.CancelledError + + class _ProfileRun(runtime.JobRun): """一个剖析任务在后台的一次执行（进队、停止、收尾的框架在 runtime.JobRun）。""" @@ -890,9 +924,10 @@ class _ProfileRun(runtime.JobRun): super().__init__(job_id) self.actor: str | None = None self.cx: catalog_profile.ProfileContext | None = None - #: 正在剖析的那张表：seq 和开始的时刻（进度回调据此落库） - self._seq = -1 - self._table_started = 0.0 + #: 不再开始新的表（_tables）：(原因, stop_reason)。止损时 stop_reason 是 failed / slow，数据源停用、关掉了剖析时为 None + self._halted: tuple[str | None, str | None] | None = None + #: 同时剖析的另一张表让整个任务出错时（_tables），正在做的几张表停下时的说法 + self._abort: str | None = None #: 开始时按此刻的目录一张表都挑不出来（_rescope）：为什么，记在任务的提醒里 self._scope_note: str | None = None #: 这个任务是怎么建的（trigger_of）：取值变化的 detected_by 按它记 @@ -910,10 +945,8 @@ class _ProfileRun(runtime.JobRun): await self._end("failed", reason, "failed", reason) async def _crashed(self, e: Exception) -> None: - from app.core.errors import describe_exception, first_line - logger.exception("数据剖析任务 %s 出错", self.job_id) - reason = JOB_ERROR.format(line=first_line(e) or describe_exception(e)) + reason = _job_error(e) await self._end("failed", reason, "failed", reason) async def _end(self, job_status: str, job_error: str | None, task_status: str, task_error: str | None) -> None: @@ -937,6 +970,8 @@ class _ProfileRun(runtime.JobRun): await source_changes.notify_after_job(job.source_id, self.job_id) def stop_text(self) -> str: + if self._abort is not None: + return self._abort return STOPPED_ERROR if self.stop_reason == "user" else INTERRUPTED_ERROR # ---------------------------------------------------------------- 主流程 @@ -971,41 +1006,82 @@ class _ProfileRun(runtime.JobRun): 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, regular, watched in pending: - # 每张表开始前重读数据源：停用了、关掉了剖析就停下；遮罩按最新的设置 - stop = await self._write(self._recheck) - if stop is not None: - await self._write(self._halt, stop, None) - break - self._table_started = time.monotonic() - await self._write(self._start_table, seq) - - async def recorded(session: AsyncSession, result: TableResult, seq: int = seq) -> None: - await self._save_table(session, seq, result) - - hooks.recorded = recorded - self._seq = seq - try: - 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: - # 任务级止损：后面的表 cancelled，带同一句原因 - await self._write(self._halt, self.cx.runner.error, self.cx.runner.stopped) - break + width = min(max(1, int(settings.table_concurrency)), catalog_profile.MAX_TABLE_CONCURRENCY) + await self._tables(pending, width) + if (halt := self._stopping()) is not None: + await self._write(self._halt, *halt) await self._settle() await self._write(self._finish) await self._notify() - # ---------------------------------------------------------------- 进度回调（引擎调） + async def _tables(self, pending: list[tuple[int, str, bool, bool]], width: int) -> None: + """按顺序剖析这些表，同时最多 width 张（设置 table_concurrency）：一张结束就开始下一张，开始的先后照任务里的顺序。 - async def _progress(self, result: TableResult) -> None: - await self._write(self._save_progress, self._seq, result) + - 每张表开始前重读数据源：停用了、关掉了剖析就不再开始新的表；止损（cx.runner）停下时同样。正在做的几张照常做完 + （止损时它们剩下的项记止损的原因），都结束之后由 _run 把还没开始的表记停下（_halt）。 + - 停止、关停：正在做的几张各自收尾（做完的项写进目录），都收完再把停止往上交。 + - 一张表让整个任务出错（快照被改动、数据源没了、记结果时写库出错）：正在做的另几张按停止收尾、说法写这个原因， + 再把错往上交，由 JobRun 记任务失败。""" + queue = collections.deque(pending) - async def _queried(self, result: TableResult) -> None: - await self._write(self._save_queries, self._seq, result) + async def worker() -> None: + while queue and self._stopping() is None: + seq, name, regular, watched = queue.popleft() + stop = await self._write(self._recheck) + if stop is not None and self._halted is None: + self._halted = (stop, None) + if self._stopping() is not None: + return # 这张没开始：还排着，由 _halt 记停下 + await self._table(seq, name, regular, watched) + + workers = [asyncio.create_task(worker(), name=f"catalog-profile-{self.job_id}-{i}") + for i in range(min(width, len(pending)))] + if not workers: + return + try: + done, rest = await asyncio.wait(workers, return_when=asyncio.FIRST_EXCEPTION) + except asyncio.CancelledError: + await _drain(workers) + raise + errors = [e for w in done if not w.cancelled() and (e := w.exception()) is not None] + if not errors: + return + from app.cluster import fence + + # 几张同时出错时按头一个说；本进程续不上租约的（Fenced）先交：那时谁都不该再写 + failed = next((e for e in errors if isinstance(e, fence.Fenced)), errors[0]) + if not isinstance(failed, fence.Fenced): + self._abort = str(failed) if isinstance(failed, runtime.JobFailed) else _job_error(failed) + await _drain(rest) + raise failed + + def _stopping(self) -> tuple[str | None, str | None] | None: + """不再开始新的表的原因（见 _halted），先到的为准：数据源停用、关掉了剖析（_tables 里重读时记下），或者止损。""" + if self._halted is None and self.cx is not None and self.cx.runner.stopped: + self._halted = (self.cx.runner.error, self.cx.runner.stopped) + return self._halted + + async def _table(self, seq: int, name: str, regular: bool, watched: bool) -> None: + """剖析一张表：检查项进度、发过的查询随时落在这张表上，整张表的结果和目录在同一个事务里写。""" + assert self.cx is not None + started = time.monotonic() + await self._write(self._start_table, seq) + + async def progress(result: TableResult) -> None: + await self._write(self._save_progress, seq, result, started) + + async def queried(result: TableResult) -> None: + await self._write(self._save_queries, seq, result, started) + + async def recorded(session: AsyncSession, result: TableResult) -> None: + await self._save_table(session, seq, result) + + hooks = TableHooks(progress=progress, query=queried, recorded=recorded, stop_text=self.stop_text) + try: + 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 # ---------------------------------------------------------------- 库操作 @@ -1093,14 +1169,12 @@ class _ProfileRun(runtime.JobRun): .values(status="running", started_at=utcnow(), snapshot_id=snapshot)) await session.commit() - def _elapsed(self) -> int: - return int((time.monotonic() - self._table_started) * 1000) - - async def _save_progress(self, session: AsyncSession, seq: int, result: TableResult) -> None: - """一项检查开始或结束：这张表的检查项、正在做哪一项、查询数、发过的查询、行数估算，任务的查询计数。""" + async def _save_progress(self, session: AsyncSession, seq: int, result: TableResult, started: float) -> None: + """一项检查开始或结束：这张表的检查项、正在做哪一项、查询数、发过的查询、行数估算，任务的查询计数。started 是这张 + 表开始的时刻（time.monotonic）。""" await session.execute(self._task(CatalogProfileTask.seq == seq, CatalogProfileTask.status == "running") .values(checks=[dict(c) for c in result.checks] or None, current=result.current, - queries=result.queries, elapsed_ms=self._elapsed(), + queries=result.queries, elapsed_ms=_elapsed(started), query_log=list(result.query_log) or None, row_estimate=result.row_estimate.to_dict() if result.row_estimate else None)) if values := self._counters(): @@ -1108,10 +1182,10 @@ class _ProfileRun(runtime.JobRun): .execution_options(synchronize_session=False)) await session.commit() - async def _save_queries(self, session: AsyncSession, seq: int, result: TableResult) -> None: + async def _save_queries(self, session: AsyncSession, seq: int, result: TableResult, started: float) -> None: """发出一条查询之后：这张表和任务的查询计数、提醒（发过的查询本身在检查项结束时一起写）。""" await session.execute(self._task(CatalogProfileTask.seq == seq, CatalogProfileTask.status == "running") - .values(queries=result.queries, elapsed_ms=self._elapsed())) + .values(queries=result.queries, elapsed_ms=_elapsed(started))) if values := self._counters(): await session.execute(update(CatalogProfileJob).where(CatalogProfileJob.id == self.job_id).values(**values) .execution_options(synchronize_session=False)) diff --git a/backend/tests/test_catalog_profile.py b/backend/tests/test_catalog_profile.py index 0a69de0a..9431d14f 100644 --- a/backend/tests/test_catalog_profile.py +++ b/backend/tests/test_catalog_profile.py @@ -201,6 +201,10 @@ async def test_settings_are_saved_and_validated_through_datasource_update(client assert "每张表时长上限" in r.json()["detail"] r = await client.patch(f"/api/datasources/{sid}", json={"options": {PROFILE_OPTION: {"enabled": "是"}}}) assert r.status_code == 422 and (r.json()["code"], r.json()["field"]) == ("profile_settings_invalid", "enabled") + r = await client.patch(f"/api/datasources/{sid}", json={"options": _on(table_concurrency=4)}) + assert r.status_code == 422 and (r.json()["code"], r.json()["field"]) == ("profile_settings_invalid", + "table_concurrency") + assert "同时剖析的表数" in r.json()["detail"] and "1 到 3 之间的整数" in r.json()["detail"] r = await client.patch(f"/api/datasources/{sid}", json={"options": _on(rows=5)}) assert r.status_code == 422 and r.json()["code"] == "profile_settings_invalid" and "field" not in r.json() r = await client.patch(f"/api/datasources/{sid}", json={"options": _on(table_max_queries=30, sample_size=500)}) @@ -209,7 +213,7 @@ async def test_settings_are_saved_and_validated_through_datasource_update(client # 设好以后就能剖析；任务里记着这次实际用的设置（快照） job = await _profile(client, sid, ["channels"]) assert job["settings"] == {"enabled": True, "table_max_queries": 30, "table_max_s": 60, "query_timeout_s": 10, - "sample_size": 500, "max_scan_rows": 100_000} + "sample_size": 500, "max_scan_rows": 100_000, "table_concurrency": 1} # ---------------------------------------------------------------- 剖析的结论 @@ -524,7 +528,7 @@ async def test_second_profile_job_on_same_source_returns_409(client, make_source second = await _start(client, sid, ["visits"]) assert second.status_code == 409 and second.json()["code"] == "profile_busy" # 说清是谁在跑、跑到哪；只点了一张表时说这张表在不在那个任务里、排第几 - assert "正在剖析：第 1 / 2 张表" in second.json()["detail"] and "visits 在这次剖析里，排第 1 张" in second.json()["detail"] + assert "正在剖析：已完成 0 / 2 张表" in second.json()["detail"] and "visits 在这次剖析里，排第 1 张" in second.json()["detail"] other = await _start(client, sid, ["gates"]) assert other.status_code == 409 and "gates 不在这次剖析里" in other.json()["detail"] active = (await client.get(f"{_url(sid)}/active")).json()["job"] diff --git a/backend/tests/test_catalog_profile_jobs.py b/backend/tests/test_catalog_profile_jobs.py index c1e6c064..db0f31f3 100644 --- a/backend/tests/test_catalog_profile_jobs.py +++ b/backend/tests/test_catalog_profile_jobs.py @@ -8,6 +8,8 @@ - 服务重启标中断、不续跑；关停时同样；重试只带没做完的表 - 任务级止损：连续 3 条失败、最近 10 条 7 条超时（停下、后面的表带同一句原因），最近 6 条 4 条超时提醒、好转后清掉 - 每张表开始前重读数据源：停用、关掉剖析就停；遮罩按最新的设置 +- 同时剖析几张表（table_concurrency）：同一时刻对业务库的查询最多这么多条、每张表仍一条一条发；几张一起显示进行中、按顺序 + 开始；停止、止损、停用时正在做的几张各自收尾；一张让任务出错时另几张照停止收尾；同一张表的行数只估一次 - 范围：需要剖析 / 全部 / 指定；空表一条查询记 skipped；视图跳过；没有可剖析的项；挑不出表 409；profile_due 的三种情形 - 留痕：IN 列表里抽到的键值不进 AgentLab 的库 - 设置迁移：旧键启动时删掉，老页面带着旧键保存只摘掉 @@ -35,7 +37,8 @@ from app.data.guard import SqlRejected from app.db.base import SessionLocal, utcnow from app.db.models import CatalogProfileJob, CatalogProfileTask, DataSource from tests.test_catalog_profile import ( # noqa: F401 - ACTOR, _detail, _finished, _job, _notes, _on, _profile, _relation, _start, _table, _url, client, make_source, + ACTOR, _detail, _finished, _job, _notes, _on, _profile, _reasons, _relation, _start, _table, _url, client, + make_source, ) @@ -525,6 +528,239 @@ async def test_masks_follow_the_latest_settings_table_by_table(client, make_sour assert not any("channel_type" in q["sql"].lower() for q in channels["query_log"]) +# -------------------------------------------------------------------------- +# 同时剖析几张表 +# -------------------------------------------------------------------------- + + +class InFlight: + """记下同一时刻最多有几条查询在业务库上、每个执行者（一张表在一个执行者里做完）同时最多几条。每条稍等一下，让同时 + 剖析的几张表真能叠在一起。""" + + def __init__(self, real) -> None: + self.real = real + self.now = self.most = 0 + self.by_runner: dict[int, int] = {} + self.most_per_runner = 0 + + async def __call__(self, source, sql, *, limits=None): + me = id(asyncio.current_task()) + self.now += 1 + self.by_runner[me] = self.by_runner.get(me, 0) + 1 + self.most = max(self.most, self.now) + self.most_per_runner = max(self.most_per_runner, self.by_runner[me]) + try: + await asyncio.sleep(0.005) + return await self.real(source, sql, limits=limits) + finally: + self.now -= 1 + self.by_runner[me] -= 1 + + +class Hold: + """查询等放行才发（when 说哪些要等）；waiting 是此刻在等的条数。""" + + def __init__(self, real, when=lambda source, sql, **kw: True) -> None: + self.real, self.when = real, when + self.waiting = 0 + self.release = asyncio.Event() + + async def __call__(self, source, sql, *, limits=None): + if self.when(source, sql, limits=limits): + self.waiting += 1 + try: + await self.release.wait() + finally: + self.waiting -= 1 + return await self.real(source, sql, limits=limits) + + +def _hold(monkeypatch, when=lambda source, sql, **kw: True) -> Hold: + hold = Hold(data_engine.run_query, when) + monkeypatch.setattr(data_engine, "run_query", hold) + return hold + + +async def _waiting(hold: Hold, n: int) -> None: + async def enough() -> bool: + return hold.waiting >= n + + await until(enough) + + +@pytest.mark.parametrize("width", [1, 2, 3]) +async def test_queries_in_flight_never_exceed_the_tables_side_by_side(client, make_source, scenic_db, monkeypatch, + width): + sid = await make_source(scenic_db, options=_on(table_concurrency=width), draft=["visits", "orders"]) + flight = InFlight(data_engine.run_query) + monkeypatch.setattr(data_engine, "run_query", flight) + job = await _profile(client, sid, ["visits", "orders", "channels", "members", "gates"]) + assert job["status"] == "succeeded" and job["settings"]["table_concurrency"] == width + assert flight.most == width, "同一时刻对业务库的查询最多同时剖析的表数那么多条" + assert flight.most_per_runner == 1, "每张表的查询仍一条一条发" + assert job["queries"] == sum(t["queries"] for t in job["tables"]) + + +async def test_tables_side_by_side_show_as_running_together_and_start_in_order(client, make_source, scenic_db, + monkeypatch): + sid = await make_source(scenic_db, options=_on(table_concurrency=2)) + hold = _hold(monkeypatch) + job = (await _start(client, sid, ["channels", "visits", "orders"])).json()["job"] + await _waiting(hold, 2) + await asyncio.sleep(0.1) + assert hold.waiting == 2, "第三张等前两张中有一张做完才开始" + running = await _job(client, sid, job["id"]) + by = {t["table_name"]: t for t in running["tasks"]} + assert by["channels"]["status"] == by["visits"]["status"] == "running" + assert by["channels"]["current"] and by["visits"]["current"] + assert by["orders"]["status"] == "pending" + busy = await _start(client, sid, ["orders"]) + assert busy.status_code == 409 and "正在剖析：已完成 0 / 3 张表" in busy.json()["detail"] + assert "orders 在这次剖析里，排第 3 张" in busy.json()["detail"] + + hold.release.set() + done = await _finished(client, sid, job["id"]) + assert done["status"] == "succeeded" + by = {t["table_name"]: t for t in done["tasks"]} + assert by["orders"]["started_at"] >= min(by["channels"]["finished_at"], by["visits"]["finished_at"]) + + +def _at_orders_codes(source, sql, **kw) -> bool: + return "GROUP BY status" in sql and "FROM orders" in sql + + +async def test_stopping_side_by_side_tables_writes_what_each_has_done(client, make_source, scenic_db, monkeypatch): + """visits、orders 同时停在码值上：两张做完的项都照常写进目录，没做的记停止；还没开始的 channels 记停止。""" + sid = await make_source(scenic_db, options=_on(table_concurrency=2), draft=["visits", "orders"]) + hold = _hold(monkeypatch, lambda source, sql, **kw: _at_visits_codes(source, sql) or _at_orders_codes(source, sql)) + job = (await _start(client, sid, ["visits", "orders", "channels"])).json()["job"] + await _waiting(hold, 2) + + r = await client.post(f"{_url(sid)}/{job['id']}/cancel") + assert r.status_code == 200, r.text + stopped = r.json()["job"] + by = {t["table_name"]: t for t in stopped["tasks"]} + assert stopped["status"] == "cancelled" + assert by["visits"]["status"] == by["orders"]["status"] == "partial" + assert by["visits"]["error"] == "剖析已停止：已做完的 4 项照常写进了数据目录，剩下 2 项没查" + assert by["orders"]["error"].startswith("剖析已停止：已做完的 ") + assert by["channels"]["status"] == "cancelled" and by["channels"]["error"] is None + assert _relation((await _notes(client, sid, "visits"))["notes"], "gate_id")["status"] == "verified" + orders = (await _notes(client, sid, "orders"))["notes"] + assert _relation(orders, "channel_id")["source"] == "fk" and _relation(orders, "channel_id")["cardinality"] + + +async def test_three_failures_stop_side_by_side_tables_and_nothing_new_starts(client, make_source, scenic_db, + monkeypatch): + """行数照常估出来，之后的查询等 visits、orders 各发出一条才一起失败；第 3 条失败时两张都还有要查的项：两张都没做完， + 剩下的项记止损的原因；后面的表不再开始。""" + sid = await make_source(scenic_db, options=_on(table_concurrency=2), draft=["visits", "orders"]) + real, sent, both = data_engine.run_query, 0, asyncio.Event() + + async def broken(source, sql, *, limits=None): + nonlocal sent + if " AS one FROM " in sql: # 估行数 + return await real(source, sql, limits=limits) + sent += 1 + if sent == 2: + both.set() + await both.wait() + raise RuntimeError("connection refused by the server") + + monkeypatch.setattr(data_engine, "run_query", broken) + job = await _profile(client, sid, ["visits", "orders", "channels", "members"]) + assert job["status"] == "partial" and job["stop_reason"] == "failed" + assert job["error"] == "连续 3 条查询失败，已停止剖析：connection refused by the server" + assert job["errors"] == 3 and all(q["outcome"] == "error" for t in job["tables"] for q in t["query_log"][1:]) + visits, orders, channels, members = (_table(job, n) for n in ("visits", "orders", "channels", "members")) + assert visits["status"] == orders["status"] == "partial", "同时在做的两张都收尾" + assert visits["error"] == orders["error"] == job["error"] + assert "failed" in _reasons(visits) and "failed" in _reasons(orders) + assert channels["status"] == members["status"] == "cancelled" + assert channels["error"] == members["error"] == job["error"] + assert channels["queries"] == members["queries"] == 0, "止损之后不再开始新的表" + + +async def test_disabling_the_source_lets_side_by_side_tables_finish(client, make_source, scenic_db, monkeypatch): + sid = await make_source(scenic_db, options=_on(table_concurrency=2)) + hold = _hold(monkeypatch, lambda source, sql, **kw: "FROM channels" in sql or "FROM orders" in sql) + job = (await _start(client, sid, ["channels", "orders", "visits"])).json()["job"] + await _waiting(hold, 2) + async with SessionLocal() as session: + row = await session.get(DataSource, sid) + row.enabled = False + await session.commit() + hold.release.set() + done = await _finished(client, sid, job["id"]) + by = {t["table_name"]: t for t in done["tasks"]} + assert by["channels"]["status"] == by["orders"]["status"] == "ok", "正在做的几张做完" + assert by["visits"]["status"] == "cancelled" + assert "已停用" in by["visits"]["error"] and by["visits"]["error"] == done["error"] + assert done["status"] == "partial" and done["stop_reason"] is None + + +async def test_a_table_failing_the_job_stops_the_side_by_side_tables_with_its_reason(client, make_source, scenic_db, + monkeypatch): + """visits 取码值时快照被改动、任务整体失败：同时在做的 channels 照停止收尾（做完的项照常写），原因写快照这一句；还没 + 开始的 orders 记失败。""" + from app.data.engine import SnapshotTampered + + sid = await make_source(scenic_db, options=_on(table_concurrency=2), draft=["visits"]) + real_codes = catalog_profile._check_codes + said = "数据文件与登记的版本不一致，可能被修改过，已拒绝查询" + + async def tampered(cx, table, *args, **kwargs): + if table == "visits": + raise SnapshotTampered(said) + return await real_codes(cx, table, *args, **kwargs) + + monkeypatch.setattr(catalog_profile, "_check_codes", tampered) + hold = _hold(monkeypatch, lambda source, sql, **kw: "FROM channels" in sql) + job = await _profile(client, sid, ["visits", "channels", "orders"]) + assert hold.waiting == 0, "等着的那条查询停掉了" + assert job["status"] == "failed" and job["error"] == said + visits, channels, orders = (_table(job, n) for n in ("visits", "channels", "orders")) + assert visits["status"] == "failed" and visits["error"].startswith(f"{said}：已查完的 4 项没有写进数据目录") + assert channels["status"] == "partial" + assert channels["error"].startswith(f"{said}：已做完的 0 项照常写进了数据目录") + assert {s["reason"] for s in channels["skipped"]} >= {"cancelled"} + assert orders["status"] == "failed" and orders["error"] == said + + +async def test_side_by_side_tables_estimate_a_shared_table_once(monkeypatch): + """两张表同时要同一张表的行数：后来的等先来的估完、用它的结果；先来的没估成（停了），后来的自己估。""" + calls: list[str] = [] + release = asyncio.Event() + + async def estimate(run, dialect, meta, table, *, max_scan_rows): + calls.append(table) + await release.wait() + if len(calls) == 1 and fail_first: + raise catalog_profile._Stop("budget") + return catalog_profile.TableSize(rows=12, method="count"), [] + + monkeypatch.setattr(catalog_profile, "estimate_size", estimate) + source = SimpleNamespace(id="s", name="shop", kind="sqlite", options={}, + schema_cache={"tables": {"members": {"columns": [{"name": "id", "type": "INTEGER"}]}}}) + for fail_first in (False, True): + calls.clear() + release.clear() + cx = catalog_profile.make_context(source, ProfileSettings(enabled=True, table_concurrency=2), options={}) + a, b = (catalog_profile._TableCx(cx, name, catalog_profile.TableResult(table=name), + catalog_profile.TableHooks()) for name in ("visits", "orders")) + first = asyncio.ensure_future(a.size("members", a.result)) + second = asyncio.ensure_future(b.size("members", b.result)) + await asyncio.sleep(0.01) + assert calls == ["members"], "后来的在等，没有再估一遍" + release.set() + got = await asyncio.gather(first, second, return_exceptions=True) + if fail_first: + assert isinstance(got[0], catalog_profile._Stop) and got[1].rows == 12 + assert calls == ["members", "members"], "先来的没估成，后来的自己估" + else: + assert got[0].rows == got[1].rows == 12 and calls == ["members"] + + # -------------------------------------------------------------------------- # 范围 # -------------------------------------------------------------------------- diff --git a/backend/tests/test_catalog_profile_units.py b/backend/tests/test_catalog_profile_units.py index 7ce3c114..c355265f 100644 --- a/backend/tests/test_catalog_profile_units.py +++ b/backend/tests/test_catalog_profile_units.py @@ -21,7 +21,7 @@ def test_settings_default_off_and_parse_leniently(): s = profile_settings({}) assert s == ProfileSettings() assert s.enabled is False and s.table_max_queries == 40 and s.table_max_s == 60 and s.query_timeout_s == 10 - assert s.sample_size == 2000 and s.max_scan_rows == 100_000 + assert s.sample_size == 2000 and s.max_scan_rows == 100_000 and s.table_concurrency == 1 on = profile_settings({PROFILE_OPTION: {"enabled": True, "table_max_queries": "30", "sample_size": 500}}) assert on.enabled is True and on.table_max_queries == 30 and on.sample_size == 500 # 库里存着的值不对（老数据、手改过）：那一项用缺省，不因此开启，也不报错；迁移前的旧键不看 @@ -42,6 +42,9 @@ def test_settings_default_off_and_parse_leniently(): ({"query_timeout_s": 61}, "单条查询时限"), ({"sample_size": 5}, "核对键值数"), ({"max_scan_rows": -1}, "整表统计行数上限"), + ({"table_concurrency": 0}, "同时剖析的表数"), + ({"table_concurrency": 4}, "同时剖析的表数"), + ({"table_concurrency": 1.5}, "同时剖析的表数"), ({"enabled": "true"}, "开启数据剖析"), ({"table_max_queries": True}, "每张表最多查询数"), ]) @@ -57,7 +60,7 @@ def test_settings_problem_rejects_unknown_keys_and_wrong_shape(): assert profile_settings_problem({PROFILE_OPTION: None}) is None assert profile_settings_problem({PROFILE_OPTION: {"enabled": True, "table_max_queries": 200, "table_max_s": 10.5, "query_timeout_s": 1.5, "sample_size": 10, - "max_scan_rows": 0}}) is None + "max_scan_rows": 0, "table_concurrency": 3}}) is None def test_legacy_whole_run_budgets_are_stripped_not_rejected(): diff --git a/frontend/src/api/client.ts b/frontend/src/api/client.ts index b4fed479..30ed5507 100644 --- a/frontend/src/api/client.ts +++ b/frontend/src/api/client.ts @@ -522,7 +522,7 @@ export const api = { /** 把范围外的表加进探查范围：当场读结构，并记进探查范围的「显式列出」 */ addToScope: (sourceId: string, tables: string[]) => post(`${catalogBase(sourceId)}/scope/add`, { tables }), /** - * 剖析任务（后台进行：逐表对业务库发只读查询，一条一条发，每张表一结束就写进目录）。剖析会对业务库发查询，默认关闭， + * 剖析任务（后台进行：逐表对业务库发只读查询，同时剖析几张表按设置、每张表的查询一条一条发，每张表一结束就写进目录）。剖析会对业务库发查询，默认关闭， * 按数据源开启。和同一个数据源的起草任务按建立顺序排队（起草在排队、进行中时剖析建成 queued、waiting_for 为 draft） */ profileJobs: { diff --git a/frontend/src/lib/terms.ts b/frontend/src/lib/terms.ts index 990e4d6b..1c7ce5b6 100644 --- a/frontend/src/lib/terms.ts +++ b/frontend/src/lib/terms.ts @@ -2991,8 +2991,9 @@ export const CATALOG_SCOPE_TEXT = { // 术语： // - 「数据剖析」：对业务库发只读查询，核实推断的关联关系（覆盖率、基数），取状态类列的码值候选，提议业务日期。 // 动作写「剖析」，不写扫描、探测。默认关闭，按数据源开启。 -// - 剖析是后台任务：逐张表进行，每张表剖析完就写进目录；查询一条一条地发，可以随时停止。关掉对话框照常进行。 -// 目录只在整张表结束时写一次：停止、关停服务时正在做的那张来得及收尾，做完的项照常写进去；进程被杀时那张表没来得及收尾， +// - 剖析是后台任务：逐张表进行，每张表剖析完就写进目录；同时剖析几张表按设置（缺省一张），每张表的查询一条一条地发，可以 +// 随时停止。关掉对话框照常进行。同时剖析不止一张时，说法里写明「同时剖析 N 张表」：业务库同时要应付这么多条查询。 +// 目录只在整张表结束时写一次：停止、关停服务时正在做的那几张来得及收尾，做完的项照常写进去；进程被杀时那几张没来得及收尾， // 这次做完的项一项都没写进去。说停下以后写进了什么，按逐表的状态照实说（cancelledNote / interruptedNote），不一律说「已写进」。 // - 预算按表算：「每张表最多查询数」「每张表时长上限」用完只停这张表，下一张表重新计数；另有单条查询时限、核对键值数、 // 整表统计行数上限。标签和服务端校验报错里的叫法一致（catalog_profile._NUMBER_FIELDS），保存被拒时按机读码认出是哪一项。 @@ -3008,6 +3009,7 @@ export const PROFILE_FIELD_LABEL: Record<'enabled' | CatalogProfileNumberKey, st query_timeout_s: '单条查询时限', sample_size: '核对键值数', max_scan_rows: '整表统计行数上限', + table_concurrency: '同时剖析的表数', } export const PROFILE_FIELD_UNIT: Record = { @@ -3016,6 +3018,7 @@ export const PROFILE_FIELD_UNIT: Record = { query_timeout_s: '秒', sample_size: '个', max_scan_rows: '行', + table_concurrency: '张', } export const PROFILE_FIELD_HINT: Record = { @@ -3024,8 +3027,13 @@ export const PROFILE_FIELD_HINT: Record = { query_timeout_s: '超过即中断这条查询；不超过数据源自身的查询时限，也不超过这张表剩下的时长', sample_size: '核对一条关联关系时，从本表取的不同键值个数：取的是数据库先给出的那些，不是随机抽样', max_scan_rows: '不超过此行数的表才做去重计数、最小值和最大值等整表统计', + table_concurrency: '大于 1 时几张表同时剖析，业务库同一时刻要应付这么多条查询，并和这个数据源上的分析共用连接；库慢、正忙时保持 1', } +/** 查询怎么发：一次一张表时「一条一条地发」，同时剖析几张时写明张数 */ +const profilePace = (tables: number) => + (tables > 1 ? `同时剖析 ${formatNumber(tables)} 张表、每张表的查询一条一条地发` : '一条一条地发') + /** 表名点名：「visits」、「refunds」 */ const quoteNames = (names: string[]) => names.map((n) => `「${n}」`).join('、') @@ -3033,8 +3041,8 @@ export const PROFILE_TEXT = { // ---- 设置 section: '数据剖析', enableHint: '默认关闭。开启后可在数据目录中核实推断的关联关系、取状态类列的码值候选', - risk: (queries: number, perTable: number, timeout: number) => - `开启后，剖析会对这个数据源发出只读查询，一条一条地发，可随时停止：每张表最多 ${formatNumber(queries)} 条、` + risk: (queries: number, perTable: number, timeout: number, tables: number) => + `开启后，剖析会对这个数据源发出只读查询，${profilePace(tables)}，可随时停止：每张表最多 ${formatNumber(queries)} 条、` + `${formatNumber(perTable)} 秒，单条不超过 ${formatNumber(timeout)} 秒。不读取明细行，遮罩的列不取样。`, range: (min: number, max: number, unit: string, def: number) => `范围 ${formatNumber(min)}–${formatNumber(max)} ${unit}，默认 ${formatNumber(def)}`, @@ -3045,8 +3053,9 @@ export const PROFILE_TEXT = { invalid: (n: number) => `数据剖析有 ${formatNumber(n)} 项设置不正确`, // ---- 数据源卡片 chip: '数据剖析已开启', - chipHint: (queries: number, perTable: number, timeout: number) => - `数据剖析已开启：每张表最多 ${formatNumber(queries)} 条只读查询、${formatNumber(perTable)} 秒，单条不超过 ${formatNumber(timeout)} 秒`, + chipHint: (queries: number, perTable: number, timeout: number, tables: number) => + `数据剖析已开启：每张表最多 ${formatNumber(queries)} 条只读查询、${formatNumber(perTable)} 秒，单条不超过 ${formatNumber(timeout)} 秒` + + (tables > 1 ? `；同时剖析 ${formatNumber(tables)} 张表` : ''), open: '设置数据剖析', dialogTitle: (name: string) => `「${name}」的数据剖析设置`, dialogBody: '设置保存在数据源上，对之后新建的剖析生效；进行中的剖析按开始时的设置。', @@ -3087,11 +3096,15 @@ export const PROFILE_TEXT = { budgetQuery: (secs: number) => `单条查询不超过 ${formatNumber(secs)} 秒`, budgetSample: (n: number) => `每条关联关系取 ${formatNumber(n)} 个不同键值核对`, budgetScan: (n: number) => (n > 0 ? `行数超过 ${formatNumber(n)} 的表不做整表统计` : '不做整表统计'), - /** 上限的估算：表数 × 每张表的上限 */ + budgetPace: (tables: number) => (tables > 1 + ? `同时剖析 ${formatNumber(tables)} 张表，每张表的查询一条一条地发` + : '一张表做完再做下一张，查询一条一条地发'), + /** 上限的估算：查询数是表数 × 每张表的上限；时长按同时剖析的张数分几轮，每轮按每张表的时长上限 */ budgetTotal: (tables: number, queries: number, span: string) => `这次 ${formatNumber(tables)} 张表，最多约 ${formatNumber(queries)} 条查询、${span}`, editSettings: '修改设置', - readOnly: '剖析会对业务库发只读查询，一条一条地发，可随时停止；停止时已做完的项照常写进目录，没做的保留原来的结论。', + readOnly: (tables: number) => + `剖析会对业务库发只读查询，${profilePace(tables)}，可随时停止；停止时已做完的项照常写进目录，没做的保留原来的结论。`, start: '开始剖析', /** 同一个数据源有起草任务在排队、进行中：剖析排在它后面 */ queue: '排队剖析', @@ -3140,7 +3153,7 @@ export const PROFILE_TEXT = { slow: '后面的表没有剖析。可以在业务库不忙的时候重试没做完的表，或在剖析设置中调低单条查询时限、每张表最多查询数。', } as Record, /** - * 停止、服务重启以后的说明。停下时正在剖析的那张表通常来得及收尾，做完的项照常写进目录（服务端记部分完成）；lost 是没来得 + * 停止、服务重启以后的说明。停下时正在剖析的那几张表通常来得及收尾，做完的项照常写进目录（服务端记部分完成）；lost 是没来得 * 及收尾的表（进程被杀、下次启动时才标中断，或停止时服务端没在跑它）：目录只在整张表结束时和它的结果一起写，所以它这次 * 一项结论都没写进去，服务端把它记成已停止、查完的项也改记跳过。有这样的表时点名说，不能一律说「做完的项已写进数据目录」。 * 它这次可能一项都还没查完，所以说「这次的结论没有写进目录」，不说「做完的项没写进去」。 diff --git a/frontend/src/pages/DataSourcesTab.tsx b/frontend/src/pages/DataSourcesTab.tsx index a5618571..60249ed8 100644 --- a/frontend/src/pages/DataSourcesTab.tsx +++ b/frontend/src/pages/DataSourcesTab.tsx @@ -691,7 +691,7 @@ function SourceCard({ row, meta, onChange, onRemoved, onEdit, onReupload, onReci )} {profile.enabled && ( - + {PROFILE_TEXT.chip} )} diff --git a/frontend/src/pages/catalog/ProfileDialog.tsx b/frontend/src/pages/catalog/ProfileDialog.tsx index 5eba1092..3373d048 100644 --- a/frontend/src/pages/catalog/ProfileDialog.tsx +++ b/frontend/src/pages/catalog/ProfileDialog.tsx @@ -34,7 +34,7 @@ import type { ProfileStartBody } from './profileJob' // // 开始前：范围（已选 / 当前筛选 / 需要剖析的表（默认）/ 全部表 / 取值观察，个数按清单上每行的 profile_due、profile_checks、 // 有没有数据目录算；取值观察只观察不剖析，最近观察过的服务端这次不观察，张数是上限）、按表写的预算和这次最多约几条查询、 -// 醒目一句「只读查询、一条一条发、可随时停止」；同一个数据源有起草任务时按钮写「排队剖析」。 +// 醒目一句「只读查询、一条一条发（同时剖析几张表时写明张数）、可随时停止」；同一个数据源有起草任务时按钮写「排队剖析」。 // 进行中：几张 / 共几张、已发几条查询、用时，按表数的真进度条，停止；逐表一行（状态、各类检查的进度、查询数、用时、正在做 // 哪一项），点一行展开这张表的检查清单、发现和发过的查询。 // 结束后：汇总（各状态几张、核实了什么）、止损和停止的原因照实写、按「有发现 / 有跳过 / 失败」筛、重试没做完的表；码值的 @@ -288,7 +288,7 @@ export function ProfileDialog({ {settings.enabled && }

- {PT.readOnly} + {PT.readOnly(settings.table_concurrency)}

{draftActive && settings.enabled && (

@@ -359,12 +359,16 @@ export function ProfileDialog({ ) } -/** 确认前写明预算：按表的查询数和时长、单条时限、抽样规模、整表统计的门槛，再按这次的表数估一个上限 */ +/** + * 确认前写明预算：按表的查询数和时长、单条时限、抽样规模、整表统计的门槛、同时剖析几张表，再按这次的表数估一个上限 + * （时长按同时剖析的张数分几轮算） + */ function Budget({ settings: s, sourceName, tables, onSettings }: { settings: CatalogProfileSettings; sourceName: string; tables: number; onSettings: () => void }) { const lines = [PT.budgetPerTable(s.table_max_queries, s.table_max_s), PT.budgetQuery(s.query_timeout_s), PT.budgetSample(s.sample_size), - PT.budgetScan(s.max_scan_rows)] + PT.budgetScan(s.max_scan_rows), PT.budgetPace(s.table_concurrency)] + const rounds = Math.ceil(tables / Math.max(1, s.table_concurrency)) return (

@@ -384,7 +388,7 @@ function Budget({ settings: s, sourceName, tables, onSettings }: { {tables > 0 && (

- {PT.budgetTotal(tables, tables * s.table_max_queries, spanText(tables * s.table_max_s))} + {PT.budgetTotal(tables, tables * s.table_max_queries, spanText(rounds * s.table_max_s))}

)}
@@ -588,7 +592,7 @@ export function checksText(t: Pick): st /** * 没来得及收尾的表：开始剖析了（有开始时刻），却以「已停止」结束。正常收尾的表状态按逐表算（完成、部分完成、失败、跳过）， - * 不会是已停止；没开始就停下的表没有开始时刻。只有进程被杀（下次启动时才标中断）或停止时服务端没在跑它，正在做的那张才以 + * 不会是已停止；没开始就停下的表没有开始时刻。只有进程被杀（下次启动时才标中断）或停止时服务端没在跑它，正在做的那几张才以 * 已停止结束——目录只在整张表结束时和它的结果一起写，它这次一项结论都没写进去（服务端把查完的项改记跳过，这张表的原因 * 写明查完的几项没写进目录；一项都还没查完时原因就是任务的那一句） */ diff --git a/frontend/src/pages/catalog/ProfileSettings.tsx b/frontend/src/pages/catalog/ProfileSettings.tsx index 91f13ee1..9baa0a4a 100644 --- a/frontend/src/pages/catalog/ProfileSettings.tsx +++ b/frontend/src/pages/catalog/ProfileSettings.tsx @@ -11,11 +11,11 @@ import { import type { ProfileForm } from './profile' // =========================================================================== -// 数据剖析的设置：开关、五项预算、风险说明。数据源编辑框里是其中一段；导入表格的源没有编辑框，数据目录页在 +// 数据剖析的设置：开关、六项预算、风险说明。数据源编辑框里是其中一段；导入表格的源没有编辑框，数据目录页在 // 剖析被拒（未开启）时也要就地打开它，所以另有一个只改这一项的弹窗（options 的其余键原样带上）。 -// 剖析会对业务库发查询，开关默认关着；开启后五项预算才展开，风险说明写的是此刻填的数。 +// 剖析会对业务库发查询，开关默认关着；开启后六项预算才展开，风险说明写的是此刻填的数。 // 预算按表算：每张表最多查询数、每张表时长上限（用完只停这张表），另有单条查询时限、核对键值数、整表统计行数上限； -// 没有整次剖析的总量上限（任务自己在连续失败、多数超时时停下）。 +// 没有整次剖析的总量上限（任务自己在连续失败、多数超时时停下）；最后一项是同时剖析几张表（缺省 1）。 // =========================================================================== /** 保存被拒时的落点：某一项，或者整段（认不出是哪一项） */ @@ -59,7 +59,7 @@ export function ProfileSettingsSection({ idPrefix, form, onChange, serverError }

- {PT.risk(s.table_max_queries, s.table_max_s, s.query_timeout_s)} + {PT.risk(s.table_max_queries, s.table_max_s, s.query_timeout_s, s.table_concurrency)}

)}
diff --git a/frontend/src/pages/catalog/profile.ts b/frontend/src/pages/catalog/profile.ts index 3c502116..bc8657d7 100644 --- a/frontend/src/pages/catalog/profile.ts +++ b/frontend/src/pages/catalog/profile.ts @@ -9,21 +9,23 @@ import { formatTime, parseServerTime } from '../../lib/format' // 规则跟着服务端（backend/app/data/catalog_profile.py 的 ProfileSettings、profile_settings、 // profile_settings_problem）走：缺省即关闭；读的时候宽松（某一项不对就用缺省），写的时候严格（超出范围的拒收）。 // -// 预算按表算（每张表最多查询数、每张表时长上限），没有整次的总量上限。以前的「查询次数上限」「总时长上限」按整次剖析算， +// 预算按表算（每张表最多查询数、每张表时长上限），没有整次的总量上限；同时剖析几张表另有一项（缺省 1）。 +// 以前的「查询次数上限」「总时长上限」按整次剖析算， // 含义不同：服务端启动时把这两个旧键删掉、保存时摘掉，这里读的时候也不认它们（表单里没有、保存时不带）。 // =========================================================================== /** 数据源 options 里剖析设置的键 */ export const PROFILE_OPTION = 'catalog_profile' -/** 设置里的数值项，也是表单上的顺序：先是按表的两项预算，再是单条查询和抽样 */ +/** 设置里的数值项，也是表单上的顺序：先是按表的两项预算，再是单条查询和抽样，最后是同时剖析几张表 */ export const PROFILE_NUMBER_KEYS: readonly CatalogProfileNumberKey[] = [ - 'table_max_queries', 'table_max_s', 'query_timeout_s', 'sample_size', 'max_scan_rows', + 'table_max_queries', 'table_max_s', 'query_timeout_s', 'sample_size', 'max_scan_rows', 'table_concurrency', ] /** 缺省值（服务端 ProfileSettings 的默认） */ export const PROFILE_DEFAULTS: CatalogProfileSettings = { enabled: false, table_max_queries: 40, table_max_s: 60, query_timeout_s: 10, sample_size: 2000, max_scan_rows: 100_000, + table_concurrency: 1, } /** 各数值项的范围（服务端 _NUMBER_FIELDS）：下限、上限、是否必须是整数 */ @@ -33,6 +35,7 @@ export const PROFILE_LIMITS: Record [ ] const NAMES = [NAME1, NAME2, NAME3] -const DEFAULTS = { enabled: false, table_max_queries: 40, table_max_s: 60, query_timeout_s: 10, sample_size: 2000, max_scan_rows: 100000 } +const DEFAULTS = { enabled: false, table_max_queries: 40, table_max_s: 60, query_timeout_s: 10, sample_size: 2000, max_scan_rows: 100000, + table_concurrency: 1 } const settingsOf = (state, src) => { const raw = state.sources.find((s) => s.id === src)?.options?.catalog_profile ?? {} const out = { ...DEFAULTS, enabled: raw.enabled === true } @@ -717,13 +718,16 @@ await section('剖析 · 设置：每张表的两项预算、被拒时报错落 && await sec.locator('[data-profile-field]').count() === 0) await sec.locator('[data-profile-enabled]').check() const labels = await sec.locator('label.label').allInnerTexts() - check('开启后五项预算展开：每张表最多查询数、每张表时长上限在前，没有整次的查询次数上限、总时长上限', - labels.join('、') === '每张表最多查询数、每张表时长上限、单条查询时限、核对键值数、整表统计行数上限', labels.join('、')) + check('开启后六项预算展开：每张表最多查询数、每张表时长上限在前，同时剖析的表数在最后，没有整次的查询次数上限、总时长上限', + labels.join('、') === '每张表最多查询数、每张表时长上限、单条查询时限、核对键值数、整表统计行数上限、同时剖析的表数', labels.join('、')) const units = await sec.locator('[data-profile-field] span').allInnerTexts() - check('……每项带单位', units.join(',') === '条,秒,秒,个,行', units.join(',')) + check('……每项带单位', units.join(',') === '条,秒,秒,个,行,张', units.join(',')) const hints = await sec.locator('[id$="-hint"]').allInnerTexts() check('……写明范围和默认值，说清用完只停这张表', hints.some((h) => h.startsWith('范围 1–200 条，默认 40') && h.includes('用完只停这张表')) && hints.some((h) => h.startsWith('范围 10–600 秒，默认 60') && h.includes('只停这张表')), hints.join(' | ')) + check('……同时剖析的表数写明范围 1–3、默认 1，说清业务库同时要应付几条查询、和分析共用连接', + hints.some((h) => h.startsWith('范围 1–3 张，默认 1') && h.includes('业务库同一时刻要应付这么多条查询') && h.includes('和这个数据源上的分析共用连接')), + hints.join(' | ')) check('……旧的两个键不回显成输入框的值', await page.locator('#ds-profile-table_max_queries').inputValue() === '' && await page.locator('#ds-profile-table_max_s').inputValue() === '') const risk = () => sec.locator('[data-profile-risk]').innerText().catch(() => '') @@ -731,6 +735,14 @@ await section('剖析 · 设置：每张表的两项预算、被拒时报错落 (await risk()).includes('一条一条地发，可随时停止：每张表最多 40 条、60 秒，单条不超过 10 秒'), await risk()) await page.locator('#ds-profile-table_max_queries').fill('30') check('……风险说明跟着填的数走', (await risk()).includes('每张表最多 30 条'), await risk()) + await page.locator('#ds-profile-table_concurrency').fill('2') + check('……同时剖析 2 张：风险说明写明张数，每张表仍一条一条地发', (await risk()).includes('只读查询，同时剖析 2 张表、每张表的查询一条一条地发，可随时停止'), + await risk()) + await page.locator('#ds-profile-table_concurrency').fill('4') + const tooMany = await page.locator('#ds-profile-table_concurrency-error').innerText().catch(() => '') + check('……填 4 张（超过 3）：就地写明范围', tooMany === '请填写 1 到 3 之间的整数', tooMany) + await page.locator('#ds-profile-table_concurrency').fill('') + check('……清空回到缺省（一张），风险说明回到「一条一条地发」', (await risk()).includes('只读查询，一条一条地发，可随时停止'), await risk()) await page.locator('#ds-profile-table_max_queries').fill('500') const err = await page.locator('#ds-profile-table_max_queries-error').innerText().catch(() => '') check('填错（500 条，超过 200）：就地写明范围，保存按钮禁用', err === '请填写 1 到 200 之间的整数' && await page.locator('[data-source-save]').isDisabled(), err) @@ -811,7 +823,8 @@ await section('剖析 · 开始前：范围、个数与预算', async () => { && (await inBox(page, '[data-profile-scope="all"]')).includes('不含视图')) const budget = await inBox(page, '[data-profile-budget]') check('预算按表写：每张表最多 40 条、60 秒，单条 10 秒，核对键值数、整表统计门槛，按哪个源的设置', - ['每张表最多 40 条查询、60 秒，用完只停这张表', '单条查询不超过 10 秒', '取 2,000 个不同键值核对', '行数超过 100,000 的表不做整表统计', `按「${NAME1}」的数据剖析设置`] + ['每张表最多 40 条查询、60 秒，用完只停这张表', '单条查询不超过 10 秒', '取 2,000 个不同键值核对', '行数超过 100,000 的表不做整表统计', + '一张表做完再做下一张，查询一条一条地发', `按「${NAME1}」的数据剖析设置`] .every((x) => budget.includes(x)), budget) check('……按这次的表数估上限：「这次 5 张表，最多约 200 条查询、5 分钟」', (await inBox(page, '[data-profile-budget-total]')) === '这次 5 张表，最多约 200 条查询、5 分钟', await inBox(page, '[data-profile-budget-total]')) @@ -1023,6 +1036,42 @@ await section('剖析 · 进行中：逐表进度、后台继续与停止', asyn await close() }) +await section('剖析 · 同时剖析几张表：预算、说法与进行中的几行', async () => { + const state = freshState() + state.sources = state.sources.map((s) => (s.id === S1 ? { ...s, options: { ...s.options, catalog_profile: { ...PROFILE_ON, table_concurrency: 2 } } } : s)) + const { page, sent, errors, close } = await open(`/data/catalog/${S1}`, { handlers: handlers(state) }) + await page.locator('[data-catalog-row]').first().waitFor() + await openDialog(page) + const budget = await inBox(page, '[data-profile-budget]') + check('预算多写一行「同时剖析 2 张表，每张表的查询一条一条地发」', budget.includes('同时剖析 2 张表，每张表的查询一条一条地发'), budget) + check('……估算：查询数照表数算，时长按两张一轮算（5 张分 3 轮）：「这次 5 张表，最多约 200 条查询、3 分钟」', + (await inBox(page, '[data-profile-budget-total]')) === '这次 5 张表，最多约 200 条查询、3 分钟', await inBox(page, '[data-profile-budget-total]')) + check('……醒目一句写明同时剖析 2 张表', (await inBox(page, '[data-profile-readonly]')).includes('只读查询，同时剖析 2 张表、每张表的查询一条一条地发，可随时停止'), + await inBox(page, '[data-profile-readonly]')) + await page.locator('[data-profile-start]').click() + await until(async () => (await phase(page)) === 'running') + const job = state.pjobs.at(-1) + check('任务记着这次的设置：同时剖析 2 张', job.settings.table_concurrency === 2, JSON.stringify(job.settings)) + settle(state, job, 'orders', { status: 'running', done: { relation: 1, codes: 0, date: 0 }, queries: 6, elapsed_ms: 4000, + current: '正在核对 orders.customer_id → customers.id' }) + settle(state, job, 'visits', { status: 'running', done: { relation: 0 }, queries: 2, elapsed_ms: 3000, current: '正在估算 visits 的行数' }) + check('进行中：两张表一起写「剖析中」，各写各正在做的那一项', await until(async () => + (await inBox(page, '[data-profile-task="orders"] [data-profile-task-status]')) === '剖析中' + && (await inBox(page, '[data-profile-task="visits"] [data-profile-task-status]')) === '剖析中') + && (await inBox(page, '[data-profile-task="orders"] [data-profile-task-note]')) === '正在核对 orders.customer_id → customers.id' + && (await inBox(page, '[data-profile-task="visits"] [data-profile-task-note]')) === '正在估算 visits 的行数', + `${await inBox(page, '[data-profile-task="orders"]')} | ${await inBox(page, '[data-profile-task="visits"]')}`) + check('……别的表还写「排队」，头部仍是做完几张 / 共几张、已发的查询合起来算', (await inBox(page, '[data-profile-task="products"] [data-profile-task-status]')) === '排队' + && (await inBox(page, '[data-profile-progress]')) === '0 / 5 张表' && (await inBox(page, '[data-profile-queries]')).includes('已发 8 条查询'), + await textOf(box(page).locator('[role="status"]'))) + await page.keyboard.press('Escape') + check('表清单：两张都写「剖析中」', await until(async () => (await rowProfile(page, 'orders')) === '剖析中' && (await rowProfile(page, 'visits')) === '剖析中'), + `${await rowProfile(page, 'orders')} / ${await rowProfile(page, 'visits')}`) + check('写请求只有建任务', writes(sent, /./).every((s) => /profile-jobs$/.test(s.key)), JSON.stringify(writes(sent, /./).map((s) => s.key))) + check('没有运行时报错', errors.length === 0, errors[0] ?? '') + await close() +}) + await section('剖析 · 结束后：汇总、筛选、展开与重试', async () => { const state = freshState() const replies = { retry: [] }