mirror of
https://github.com/ThearchyHelios/AgentLab.git
synced 2026-11-15 16:01:59 +00:00
剖析的范围在开始时定:排在起草后面的剖析任务,开始时按起草后的目录重挑要剖析的表
真实后端验收里「起草完接着剖析」这条主路走不通:剖析任务建的那一刻就按当时的目录定了表单,可它排在起草后面,起草还没提出 关系,只靠关系才有检查项的表(链接表、明细表)一张都挑不进来,起草写完关系后剖析照旧跳过它们。 - 开始剖析时(_ProfileRun._rescope)按那一刻的目录重定:需要剖析的表、全部表整份重挑(这个任务的表都还没开始);指定的表里 建任务时判为「没有可剖析的项」的再看一次,现在有了就排上。 - 排在起草后面时挑不出表也照建,不再 409(起草多半会提出关系);开始时还挑不出来,任务以 0 张表结束,提醒里写明为什么。 - 界面:排队时写明要剖析哪些表开始时按起草后的目录再定,开始前的张数是按现在的目录算的。 - 新测试在去掉重挑的代码上两条都失败。
This commit is contained in:
@@ -509,26 +509,46 @@ async def _pick(session: AsyncSession, row: Any, source: Any, *, tables: list[st
|
||||
return [(name, None, None, checks) for name, checks in candidates]
|
||||
|
||||
|
||||
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]]:
|
||||
"""建一个排队的剖析任务(还没开始,调用方接着 manager.start)。row 是数据源记录(遮罩按它的设置),source 是读表结构
|
||||
用的源(上传源是绑定当前快照的视图)。同源已有排队或进行中的剖析任务抛 ProfileBusy;挑出 0 张表抛 ProfileNothing。
|
||||
同源有起草任务时照建,排在它后面。建完顺手清掉超出保留数的旧任务。"""
|
||||
await ensure_free(session, row, tables)
|
||||
picked = await _pick(session, row, source, tables=tables, scope=scope)
|
||||
job = CatalogProfileJob(id=new_id(), source_id=row.id, status="queued", scope=scope,
|
||||
settings=settings.to_dict(), created_by=actor, retry_of=retry_of)
|
||||
now = utcnow()
|
||||
def _tasks_for(job_id: str, picked: list[tuple[str, str | None, str | None, list[Any]]],
|
||||
now: Any) -> list[CatalogProfileTask]:
|
||||
"""_pick 挑出的表 → 逐表的行:按规则不剖析的(表结构里没有、视图、没有可剖析的项)直接记跳过,其余排队。"""
|
||||
tasks: list[CatalogProfileTask] = []
|
||||
for seq, (name, reason, why, checks) in enumerate(picked):
|
||||
shown = [c.as_check() for c in checks]
|
||||
if reason is not None:
|
||||
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))
|
||||
tasks.append(_skipped_task(job_id, seq, name, reason, str(why), shown, rule, now))
|
||||
continue
|
||||
tasks.append(CatalogProfileTask(job_id=job.id, seq=seq, table_name=name, status="pending", checks=shown,
|
||||
tasks.append(CatalogProfileTask(job_id=job_id, seq=seq, table_name=name, status="pending", checks=shown,
|
||||
summary=_empty_summary()))
|
||||
return tasks
|
||||
|
||||
|
||||
async def _behind_draft(session: AsyncSession, source_id: str) -> bool:
|
||||
"""这个源有排队、进行中的起草任务,或者正在同步起草:新建的剖析任务要排在它后面。"""
|
||||
from app.data import catalog_jobs
|
||||
|
||||
return catalog_jobs.sync_drafting(source_id) or await catalog_jobs.active_job(session, source_id) is not None
|
||||
|
||||
|
||||
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]]:
|
||||
"""建一个排队的剖析任务(还没开始,调用方接着 manager.start)。row 是数据源记录(遮罩按它的设置),source 是读表结构
|
||||
用的源(上传源是绑定当前快照的视图)。同源已有排队或进行中的剖析任务抛 ProfileBusy;挑出 0 张表抛 ProfileNothing。
|
||||
同源有起草任务时照建,排在它后面——这时挑出 0 张表也照建。不管排没排队,这里挑出的表单都是暂定的:开始剖析时按那时的
|
||||
目录再定一遍(_ProfileRun._rescope)。建完顺手清掉超出保留数的旧任务。"""
|
||||
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):
|
||||
raise
|
||||
picked = []
|
||||
job = CatalogProfileJob(id=new_id(), source_id=row.id, status="queued", scope=scope,
|
||||
settings=settings.to_dict(), created_by=actor, retry_of=retry_of)
|
||||
tasks = _tasks_for(job.id, picked, utcnow())
|
||||
session.add(job)
|
||||
try:
|
||||
await session.flush()
|
||||
@@ -690,6 +710,8 @@ class _ProfileRun(runtime.JobRun):
|
||||
#: 正在剖析的那张表:seq 和开始的时刻(进度回调据此落库)
|
||||
self._seq = -1
|
||||
self._table_started = 0.0
|
||||
#: 开始时按此刻的目录一张表都挑不出来(_rescope):为什么,记在任务的提醒里
|
||||
self._scope_note: str | None = None
|
||||
|
||||
# ---------------------------------------------------------------- 停止、出错时收尾
|
||||
|
||||
@@ -738,10 +760,14 @@ class _ProfileRun(runtime.JobRun):
|
||||
source = await table_versions.resolve_source(session, row)
|
||||
except table_versions.SnapshotMissing as e:
|
||||
raise runtime.JobFailed(str(e)) from e
|
||||
pending = [(t.seq, t.table_name) for t in await task_rows(session, job.id) if t.status == "pending"]
|
||||
settings = ProfileSettings.from_dict(job.settings)
|
||||
self.actor = job.created_by
|
||||
options = dict(row.options or {})
|
||||
# 要剖析哪些表按开始这一刻的目录定:排在起草后面时,建任务那会儿起草还没提出关系
|
||||
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"]
|
||||
self.actor = job.created_by
|
||||
try:
|
||||
self.cx = catalog_profile.make_context(source, settings, options=options)
|
||||
except ValueError as e: # 不支持的数据库类型
|
||||
@@ -791,7 +817,7 @@ class _ProfileRun(runtime.JobRun):
|
||||
if runner is None:
|
||||
return {}
|
||||
return {"queries": runner.queries, "timeouts": runner.timeouts, "errors": runner.errors,
|
||||
"notice": runner.notice}
|
||||
"notice": runner.notice or self._scope_note}
|
||||
|
||||
async def _save_counters(self, session: AsyncSession) -> None:
|
||||
if values := self._counters():
|
||||
@@ -802,6 +828,40 @@ class _ProfileRun(runtime.JobRun):
|
||||
async def _claim(self, session: AsyncSession) -> bool:
|
||||
return await runtime.claim(session, CatalogProfileJob, self.job_id)
|
||||
|
||||
async def _rescope(self, session: AsyncSession, row: Any, source: Any) -> None:
|
||||
"""开始剖析时按此刻的目录重定范围(建任务时挑的表单是暂定的:排在起草后面时,起草还没提出关系)。
|
||||
|
||||
- 需要剖析的表、全部表:整份重挑(这个任务的表都还没开始)。挑不出来就照实记在提醒里,任务以 0 张表结束;
|
||||
- 指定的表:建任务时判为「没有可剖析的项」的再看一次,现在有了就排上;表结构里没有、视图照旧跳过。
|
||||
要做的检查项在每张表开始时还会按那时的目录再列一遍(profile_table)。"""
|
||||
job = await session.get(CatalogProfileJob, self.job_id)
|
||||
if job is None:
|
||||
return
|
||||
if job.scope == "selected":
|
||||
stale = [t for t in await task_rows(session, job.id) if t.status == "skipped" and t.error == NOTHING_ERROR]
|
||||
if stale:
|
||||
fresh = {name: (reason, checks) for name, reason, _why, checks in await _pick(
|
||||
session, row, source, tables=[t.table_name for t in stale], scope="selected")}
|
||||
for task in stale:
|
||||
reason, checks = fresh.get(task.table_name, ("nothing", []))
|
||||
if reason is None:
|
||||
await session.execute(self._task(CatalogProfileTask.seq == task.seq).values(
|
||||
status="pending", checks=[c.as_check() for c in checks], skipped=None, skipped_count=0,
|
||||
unfinished=0, error=None, finished_at=None))
|
||||
await session.commit()
|
||||
return
|
||||
await session.execute(delete(CatalogProfileTask).where(CatalogProfileTask.job_id == self.job_id)
|
||||
.execution_options(synchronize_session=False))
|
||||
try:
|
||||
picked = await _pick(session, row, source, tables=None, scope=job.scope)
|
||||
except ProfileNothing as e:
|
||||
picked = []
|
||||
self._scope_note = str(e)
|
||||
await session.execute(update(CatalogProfileJob).where(CatalogProfileJob.id == self.job_id)
|
||||
.values(notice=self._scope_note).execution_options(synchronize_session=False))
|
||||
session.add_all(_tasks_for(self.job_id, picked, utcnow()))
|
||||
await session.commit()
|
||||
|
||||
async def _recheck(self, session: AsyncSession) -> str | None:
|
||||
"""重读数据源:停用了、关掉了剖析返回停下的整句;遮罩换成最新的设置。数据源没了抛 JobFailed。"""
|
||||
job = await session.get(CatalogProfileJob, self.job_id)
|
||||
|
||||
@@ -838,3 +838,51 @@ async def test_cancel_a_job_claimed_between_load_and_cancel(client, make_source,
|
||||
await other.commit()
|
||||
await catalog_jobs.cancel(session, job)
|
||||
assert stopped == [job_id, draft_id]
|
||||
|
||||
|
||||
async def test_scope_is_settled_when_the_job_starts_behind_a_draft(client, make_source, scenic_db, monkeypatch):
|
||||
"""起草完接着剖析:剖析任务建的时候排在起草后面,起草还没提出关系,只靠关系才有检查项的表(areas.park_id 只能靠命名
|
||||
推断)这时挑不进来。范围在开始剖析时按那时的目录定:起草写完关系,剖析开始时 areas 排上了。"""
|
||||
sid = await make_source(scenic_db, options=_on())
|
||||
assert (await _rows(client, sid))["areas"]["profile_checks"] == 0, "还没起草:areas 没有可剖析的项"
|
||||
draft_start, profile_start = catalog_jobs.manager.start, catalog_profile_jobs.manager.start
|
||||
monkeypatch.setattr(catalog_jobs.manager, "start", lambda job_id: None)
|
||||
monkeypatch.setattr(catalog_profile_jobs.manager, "start", lambda job_id: None)
|
||||
r = await client.post(f"/api/datasources/{sid}/catalog/draft-jobs", json={"tables": ["areas"], "use_model": False})
|
||||
assert r.status_code == 202, r.text
|
||||
draft_id = r.json()["job"]["id"]
|
||||
queued = await _start(client, sid, scope="all")
|
||||
assert queued.status_code == 202, queued.text
|
||||
job = queued.json()["job"]
|
||||
assert job["waiting_for"] == "draft" and "areas" not in [t["table_name"] for t in job["tasks"]]
|
||||
|
||||
draft_start(draft_id) # 起草跑完:areas 有了推断的关系
|
||||
await catalog_jobs.manager.wait(draft_id, timeout=30)
|
||||
assert (await _rows(client, sid))["areas"]["profile_checks"] == 1
|
||||
profile_start(job["id"])
|
||||
await catalog_profile_jobs.manager.wait(job["id"], timeout=60)
|
||||
tasks = {t["table_name"]: t for t in (await client.get(f"{_url(sid)}/{job['id']}")).json()["job"]["tasks"]}
|
||||
assert "areas" in tasks and tasks["areas"]["status"] in ("ok", "partial"), tasks.get("areas")
|
||||
assert tasks["areas"]["checks_by_kind"]["relation"][1] == 1
|
||||
|
||||
|
||||
async def test_nothing_needed_behind_a_draft_is_created_and_settled_at_start(client, make_source, tmp_path,
|
||||
monkeypatch):
|
||||
"""排在起草后面时挑不出表也照建(起草多半会提出关系);开始时还是挑不出来,就以 0 张表结束、提醒里写明为什么。"""
|
||||
path = _db(tmp_path / "bare.db", "CREATE TABLE notes_kv (k TEXT PRIMARY KEY, v TEXT);", {"notes_kv": [("a", "b")]})
|
||||
sid = await make_source(path, options=_on())
|
||||
assert (await _start(client, sid)).status_code == 409, "没有起草任务时照旧 409"
|
||||
draft_start = catalog_jobs.manager.start
|
||||
monkeypatch.setattr(catalog_jobs.manager, "start", lambda job_id: None)
|
||||
r = await client.post(f"/api/datasources/{sid}/catalog/draft-jobs", json={"tables": ["notes_kv"], "use_model": False})
|
||||
draft_id = r.json()["job"]["id"]
|
||||
made = await _start(client, sid)
|
||||
assert made.status_code == 202, made.text
|
||||
job = made.json()["job"]
|
||||
assert job["status"] == "queued" and job["total"] == 0 and job["waiting_for"] == "draft"
|
||||
draft_start(draft_id)
|
||||
await catalog_jobs.manager.wait(draft_id, timeout=30)
|
||||
await catalog_profile_jobs.manager.wait(job["id"], timeout=30)
|
||||
done = (await client.get(f"{_url(sid)}/{job['id']}")).json()["job"]
|
||||
assert done["status"] == "succeeded" and done["total"] == 0
|
||||
assert done["notice"] and done["notice"].startswith("没有需要剖析的表")
|
||||
|
||||
@@ -2873,7 +2873,7 @@ export const PROFILE_TEXT = {
|
||||
start: '开始剖析',
|
||||
/** 同一个数据源有起草任务在排队、进行中:剖析排在它后面 */
|
||||
queue: '排队剖析',
|
||||
queueHint: '这个数据源有起草任务在进行:剖析会排在它后面,起草任务结束后自动开始',
|
||||
queueHint: '这个数据源有起草任务在进行:剖析会排在它后面,起草任务结束后自动开始。上面的张数按现在的目录算,开始时按起草后的目录再定',
|
||||
disabledTitle: '这个数据源未开启数据剖析',
|
||||
disabledBody: '剖析会对业务库发出只读查询,需要先在数据源设置中开启。',
|
||||
enable: '开启数据剖析',
|
||||
@@ -2892,7 +2892,7 @@ export const PROFILE_TEXT = {
|
||||
cancelled: '已停止',
|
||||
interrupted: '服务重启,剖析中断',
|
||||
} as Record<CatalogDraftJobStatus, string>,
|
||||
waitingDraft: '排在起草任务后面,起草结束后自动开始',
|
||||
waitingDraft: '排在起草任务后面,起草结束后自动开始。要剖析哪些表,开始时按起草后的目录再定',
|
||||
// 停止时服务端不等正在执行的那条查询(它的结果记「失败」),库那边按单条时限截止它;收尾写完(至多十来秒)停止就答。
|
||||
// 别写成「等这条查询结束」:单条时限设得长时,界面会让人以为要等满时限
|
||||
stopping: '正在停止:正在执行的那条查询不再等待,由数据库按单条时限截止;已做完的项照常写进目录…',
|
||||
|
||||
@@ -1350,8 +1350,8 @@ await section('剖析 · 和起草排队', async () => {
|
||||
await page.locator('[data-profile-start]').click()
|
||||
await until(async () => (await phase(page)) === 'running')
|
||||
const job = state.pjobs.at(-1)
|
||||
check('……建出来是排队,写「排在起草任务后面,起草结束后自动开始」', job.waiting_for === 'draft' && (await inBox(page, '[data-profile-title]')).startsWith('排队中')
|
||||
&& (await inBox(page, '[data-profile-waiting="draft"]')) === '排在起草任务后面,起草结束后自动开始')
|
||||
check('……建出来是排队,写「排在起草任务后面,起草结束后自动开始」、开始时按起草后的目录再定', job.waiting_for === 'draft' && (await inBox(page, '[data-profile-title]')).startsWith('排队中')
|
||||
&& (await inBox(page, '[data-profile-waiting="draft"]')) === '排在起草任务后面,起草结束后自动开始。要剖析哪些表,开始时按起草后的目录再定')
|
||||
await page.locator('[data-profile-hide]').click()
|
||||
check('……页头写「剖析 排队中」,悬停写排在起草任务后面', await until(async () => (await head(page).getAttribute('data-catalog-profile')) === 'queued')
|
||||
&& (await textOf(head(page))) === '剖析 排队中' && (await head(page).getAttribute('title')).includes('排在起草任务后面'))
|
||||
|
||||
Reference in New Issue
Block a user