From fc25c6367cbe95ae402b43bfe9c542a9a77c8ed9 Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Wed, 14 Oct 2026 14:07:14 +0800 Subject: [PATCH] =?UTF-8?q?=E8=B5=B7=E8=8D=89=E4=BB=BB=E5=8A=A1=E5=86=99?= =?UTF-8?q?=E8=BF=9B=E7=9B=AE=E5=BD=95=E7=9A=84=E9=A1=B9=EF=BC=8C=E4=BF=AE?= =?UTF-8?q?=E6=94=B9=E6=97=B6=E9=97=B4=E5=8F=96=E9=82=A3=E4=B8=80=E6=AC=A1?= =?UTF-8?q?=E5=86=99=E5=85=A5=E7=9A=84=E6=97=B6=E9=97=B4=EF=BC=9A=E5=AE=BD?= =?UTF-8?q?=E8=A1=A8=E9=80=90=E6=AE=B5=E8=90=BD=E7=9B=98=E7=9A=84=E6=AF=8F?= =?UTF-8?q?=E4=B8=80=E6=AE=B5=E3=80=81=E6=95=B4=E5=BC=A0=E8=A1=A8=E6=94=B6?= =?UTF-8?q?=E5=B0=BE=E5=90=84=E5=8F=96=E4=B8=80=E6=AC=A1=E5=BD=93=E6=97=B6?= =?UTF-8?q?=E7=9A=84=E6=97=B6=E9=97=B4=EF=BC=8C=E4=B8=8D=E5=86=8D=E6=95=B4?= =?UTF-8?q?=E4=B8=AA=E4=BB=BB=E5=8A=A1=E9=83=BD=E8=AE=B0=E6=88=90=E5=BC=80?= =?UTF-8?q?=E5=A7=8B=E7=9A=84=E9=82=A3=E4=B8=80=E5=88=BB=EF=BC=88=E4=BB=A5?= =?UTF-8?q?=E5=89=8D=E5=8F=AA=E5=9C=A8=E4=BB=BB=E5=8A=A1=E5=BC=80=E5=A7=8B?= =?UTF-8?q?=E6=97=B6=E5=8F=96=E4=B8=80=E6=AC=A1=E3=80=81=E6=88=AA=E5=88=B0?= =?UTF-8?q?=E7=A7=92=EF=BC=8C=E6=AF=94=E4=BB=BB=E5=8A=A1=E7=9A=84=E5=BC=80?= =?UTF-8?q?=E5=A7=8B=E6=97=B6=E9=97=B4=E8=BF=98=E6=97=A9=EF=BC=8C=E4=B9=9F?= =?UTF-8?q?=E7=9C=8B=E4=B8=8D=E5=87=BA=E5=93=AA=E4=B8=80=E6=AE=B5=E6=98=AF?= =?UTF-8?q?=E4=BB=80=E4=B9=88=E6=97=B6=E5=80=99=E5=86=99=E8=BF=9B=E5=8E=BB?= =?UTF-8?q?=E7=9A=84=EF=BC=89=EF=BC=9B=E8=A1=A5=E4=B8=80=E6=9D=A1=E7=94=A8?= =?UTF-8?q?=E5=81=87=E6=97=B6=E9=92=9F=E9=80=90=E6=AE=B5=E6=A0=B8=E5=AF=B9?= =?UTF-8?q?=E7=9A=84=E6=B5=8B=E8=AF=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/app/data/catalog_jobs.py | 15 ++++++++------ backend/tests/test_catalog_draft_jobs.py | 26 ++++++++++++++++++++++++ 2 files changed, 35 insertions(+), 6 deletions(-) diff --git a/backend/app/data/catalog_jobs.py b/backend/app/data/catalog_jobs.py index 48c37056..d63c0313 100644 --- a/backend/app/data/catalog_jobs.py +++ b/backend/app/data/catalog_jobs.py @@ -474,7 +474,6 @@ class _JobRun: self._started: set[str] = set() self.source: Any = None self.actor: str | None = None - self.at = "" # ---------------------------------------------------------------- 库写入 @@ -553,7 +552,7 @@ class _JobRun: except catalog.CatalogModelUnavailable as e: raise _JobFailed(str(e)) from e existing = {t: e.notes for t, e in (await catalog.read_catalog(session, source.id)).items()} - self.source, self.at = source, catalog.now_iso() + self.source = source all_tables = (getattr(source, "schema_cache", None) or {}).get("tables") or {} names = [n for n in pending if isinstance(all_tables.get(n), dict)] gone = [n for n in pending if n not in names] @@ -653,9 +652,12 @@ class _JobRun: async def _write_part(self, session: AsyncSession, table: str, written: catalog_draft.PartWrite) -> None: """宽表写出了一段：马上并入目录（这一次不删任何东西），记下写进去的段数、累加计数。写不进去只记日志：整张表结束 - 时收尾那次还会把它带上。""" + 时收尾那次还会把它带上。 + + 新增、更新的项的修改时间是这一次写入的时间（每次写入各取一次）：以前整个任务只在开始时取一次，起草了几十分钟的宽表， + 每一段写进去的项都记成任务开始那一刻，比任务的开始时间还早。""" try: - found = await write_part(session, self.source, table, written.notes, at=self.at, actor=self.actor) + found = await write_part(session, self.source, table, written.notes, at=catalog.now_iso(), actor=self.actor) except Exception: # noqa: BLE001 - 这一段写不进去不影响起草 logger.exception("起草任务 %s 把表 %s 写出来的一段并入数据目录时出错", self.job_id, table) await session.rollback() @@ -674,9 +676,10 @@ class _JobRun: async def _write_table(self, session: AsyncSession, table: str, outcome: catalog_draft.ModelOutcome | None) -> None: """一张表结束：并入目录、写入（宽表写过的段已经并入过，这里是整张表的收尾），记下这张表的结果。计数加在逐段写入 - 的计数上。""" + 的计数上。修改时间取这一次写入的时间（同 _write_part）：逐段写进去、内容没变的项留着那一段的时间。""" try: - result = await catalog.write_draft(session, self.source, table, outcome, at=self.at, actor=self.actor) + result = await catalog.write_draft(session, self.source, table, outcome, at=catalog.now_iso(), + actor=self.actor) except Exception as e: # noqa: BLE001 - 这一张写不进去不影响别的表 from app.core.errors import describe_exception, first_line diff --git a/backend/tests/test_catalog_draft_jobs.py b/backend/tests/test_catalog_draft_jobs.py index 81840b34..2ce4aef9 100644 --- a/backend/tests/test_catalog_draft_jobs.py +++ b/backend/tests/test_catalog_draft_jobs.py @@ -17,7 +17,9 @@ from __future__ import annotations import asyncio +import itertools import uuid +from datetime import datetime, timedelta, timezone from types import SimpleNamespace from urllib.parse import quote @@ -1052,3 +1054,27 @@ async def test_a_dropped_filter_is_logged_on_the_task_not_failed(client, monkeyp assert by["channels"]["log"] == [] notes = (await client.get(f"{_base(sid)}/gates")).json()["notes"] assert "valid_filter" not in notes and notes["label"]["value"] == "闸口" + + +async def test_each_write_of_a_job_stamps_its_own_time(client, monkeypatch): + """起草任务写进目录的项，修改时间是那一次写入的时间：宽表逐段落盘的每一段、整张表收尾各取一次当时的时间。以前只在任务 + 开始时取一次（截到秒），整个任务写进去的项都是开始那一刻，比任务的开始时间还早，也看不出哪一段是什么时候写的。 + + 时钟换成每取一次走一分钟的假时钟；同时只起草一段，各段的先后是定的：表级那段、第 2、3 段逐段落盘，第 4 段由收尾写进去。""" + ticks = itertools.count() + start = datetime(2026, 10, 14, 3, tzinfo=timezone.utc) + monkeypatch.setattr(catalog, "now_iso", + lambda: (start + timedelta(minutes=next(ticks))).isoformat(timespec="seconds")) + monkeypatch.setattr(catalog, "MODEL_CONCURRENCY", 1) + sid = await _source(_wide()) + job = await _draft(client, sid, monkeypatch, FakeModel(_wide_reply()), ["wide"]) + task = job["tasks"][0] + assert task["status"] == "ok" and task["parts_written"] == 4 + notes = (await client.get(f"{_base(sid)}/wide")).json()["notes"] + cols = notes["columns"] + parts = [{cols[f"c{i:03d}"]["label"]["updated_at"] for i in range(lo, hi + 1)} + for lo, hi in ((1, 32), (33, 64), (65, 95))] + assert all(len(p) == 1 for p in parts), "同一段的项是同一次写进去的" + assert notes["kind"]["updated_at"] == notes["label"]["updated_at"], "表级那段的两项是同一次写进去的" + stamps = [notes["label"]["updated_at"], *(p.pop() for p in parts)] + assert len(set(stamps)) == 4 and stamps == sorted(stamps), stamps