From aa051b0a28a86af41177f1b9d8a5b09cf028eb26 Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Mon, 19 Oct 2026 09:03:37 +0800 Subject: [PATCH] =?UTF-8?q?=E8=BF=90=E8=A1=8C=E5=AF=B9=E6=AF=94=E9=87=8C?= =?UTF-8?q?=E5=AD=90=E5=B7=A5=E4=BD=9C=E6=B5=81=E7=9A=84=E5=8F=A3=E5=BE=84?= =?UTF-8?q?=E5=8D=A1=E6=8C=89=E5=8D=A1=E6=89=80=E5=9C=A8=E9=82=A3=E4=B8=80?= =?UTF-8?q?=E5=B1=82=E6=89=BE=E4=BA=A7=E5=87=BA=EF=BC=9A=E7=BB=84=E8=A3=85?= =?UTF-8?q?=E6=8E=A8=E6=BC=94=E7=94=A8=E7=9A=84=20Sources=20=E6=97=B6?= =?UTF-8?q?=E5=B8=A6=E4=B8=8A=E6=98=AF=E5=93=AA=E5=BC=A0=E5=8D=A1=EF=BC=8C?= =?UTF-8?q?=E8=8A=82=E7=82=B9=E4=BA=A7=E5=87=BA=E3=80=81=E8=B7=91=E5=AE=8C?= =?UTF-8?q?=E7=9A=84=E6=97=B6=E5=88=BB=E6=8C=89=20engine/lineage.in=5Fscop?= =?UTF-8?q?e=20=E6=8D=A2=E6=88=90=E5=8D=A1=E9=82=A3=E4=B8=80=E5=B1=82?= =?UTF-8?q?=E7=9A=84=20id=EF=BC=88=E5=90=8C=E6=8E=A8=E6=BC=94=20api/whatif?= =?UTF-8?q?.=5Fsources=E3=80=81=E8=AF=81=E6=8D=AE=E9=9D=A2=E6=9D=BF=20=5FI?= =?UTF-8?q?nputMasks.=5Foutput=EF=BC=89=E3=80=82=E4=BB=A5=E5=89=8D?= =?UTF-8?q?=E5=AD=90=E5=9B=BE=E9=87=8C=E7=9A=84=E5=8D=A1=E5=8F=96=E5=80=BC?= =?UTF-8?q?=E9=93=BE=E5=86=99=E7=9A=84=E6=98=AF=E5=AD=90=E5=9B=BE=E9=87=8C?= =?UTF-8?q?=E7=9A=84=20id=EF=BC=88nodes.parse=EF=BC=89=EF=BC=8C=E5=AF=B9?= =?UTF-8?q?=E6=AF=94=E6=8B=BF=E7=88=B6=E5=9B=BE=E9=87=8C=E5=90=8C=E5=90=8D?= =?UTF-8?q?=E7=9A=84=E8=8A=82=E7=82=B9=E5=BD=93=E5=AE=83=E7=9A=84=E4=BA=A7?= =?UTF-8?q?=E5=87=BA=EF=BC=8C=E5=AE=B9=E5=99=A8=E8=BE=93=E5=85=A5=E8=AE=A4?= =?UTF-8?q?=E5=88=B0=E7=88=B6=E5=9B=BE=E9=82=A3=E6=AC=A1=E6=9F=A5=E8=AF=A2?= =?UTF-8?q?=E4=B8=8A=EF=BC=9B=E5=8A=A0=E4=B8=80=E6=9D=A1=E5=AD=90=E5=9B=BE?= =?UTF-8?q?=E5=8D=A1=E7=9A=84=E7=9C=9F=E8=B7=91=E5=AF=B9=E6=AF=94=E6=B5=8B?= =?UTF-8?q?=E8=AF=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/app/api/compare.py | 12 +++- backend/app/engine/compare.py | 9 +-- backend/tests/fixtures/compare_runs.py | 2 +- backend/tests/test_compare_api.py | 76 ++++++++++++++++++++++++++ 4 files changed, 91 insertions(+), 8 deletions(-) diff --git a/backend/app/api/compare.py b/backend/app/api/compare.py index 16fbf75f..aa82f7d9 100644 --- a/backend/app/api/compare.py +++ b/backend/app/api/compare.py @@ -184,9 +184,15 @@ async def _side(run: Run, sealed: _Sealed, masks: _Masks, kinds: dict[str, Any]) for artifact, info in sealed.queries.items()} returned = _returned(sealed) - def sources(started: int) -> recompute.Sources: - return recompute.Sources(load=loader, queries=tuple(sealed.queries), outputs=_outputs_before(sealed, started), - masked=masked, returned=returned, finished=_finished_before(sealed, started)) + def sources(started: int, card_node: str = "") -> recompute.Sources: + # 节点产出、跑完的时刻按口径卡所在那一层的 id 交进去（engine/lineage.in_scope，同推演 api/whatif._sources）：子工作流 + # 里的卡（「子工作流节点 id/卡」）取值链写的是子图里的 id，只看同一张子图里的节点，父图里同名的不算 + from app.engine.lineage import in_scope + + return recompute.Sources(load=loader, queries=tuple(sealed.queries), + outputs=in_scope(_outputs_before(sealed, started), card_node), + masked=masked, returned=returned, + finished=in_scope(_finished_before(sealed, started), card_node)) return engine.Side(run=run_ref(run), graph=sealed.graph, events=sealed.events, cards=cards, load=loader, masked=masked, sources=sources, dialects=kinds, inputs=dict(run.input or {})) diff --git a/backend/app/engine/compare.py b/backend/app/engine/compare.py index 49839bf1..bbc278ae 100644 --- a/backend/app/engine/compare.py +++ b/backend/app/engine/compare.py @@ -183,8 +183,9 @@ class Side: 没有的，就是读不出来或者被改过 - load：工件 id → 内容，只认封存范围内的；文件不在返回 None，内容和哈希对不上抛 ValueError - masked：查询快照 → 要遮的列（小写）：查询当时记下的加数据源现在设的 - - sources：口径卡那次执行开始的事件序号 → 推演重算还原输入要用的 Sources（节点产出、跑完的时刻都按那一刻）。 - 不给就按 load、masked 拼一个最小的：nodes.* 这类容器输入还原不出来 + - sources：(口径卡那次执行开始的事件序号, 卡在这次运行里的节点 id) → 推演重算还原输入要用的 Sources（节点产出、跑完的 + 时刻都按那一刻，按卡所在那一层的 id 记：子工作流里的卡只看同一张子图里的节点）。不给就按 load、masked 拼一个最小的： + nodes.* 这类容器输入还原不出来 - dialects：数据源名 → 类型（sqlite / postgres …），规范化 SQL 时按它的方言解析 - inputs：运行输入（Run.input） """ @@ -195,7 +196,7 @@ class Side: cards: list[tuple[str, str, dict[str, Any]]] = field(default_factory=list) load: Callable[[str], Any] = lambda _: None masked: Mapping[str, frozenset[str]] = field(default_factory=dict) - sources: Callable[[int], recompute.Sources] | None = None + sources: Callable[[int, str], recompute.Sources] | None = None dialects: Mapping[str, str | None] = field(default_factory=dict) inputs: dict[str, Any] = field(default_factory=dict) @@ -454,7 +455,7 @@ def _plan_card(side: Side, card: _Card, queries: list[_Query]) -> None: """按推演重算的规则还原卡上每个指标的输入：落在哪份快照上、是不是遮罩列、容器的完整值。""" if card.content is None: return - sources = side.sources(card.started) if side.sources is not None else recompute.Sources( + sources = side.sources(card.started, card.node_id) if side.sources is not None else recompute.Sources( load=side.load, queries=tuple(dict.fromkeys(q.artifact for q in queries)), masked=side.masked) store = recompute._Store(sources) card.sources = sources diff --git a/backend/tests/fixtures/compare_runs.py b/backend/tests/fixtures/compare_runs.py index 9e0d3944..455f28d5 100644 --- a/backend/tests/fixtures/compare_runs.py +++ b/backend/tests/fixtures/compare_runs.py @@ -187,7 +187,7 @@ class Fake: return artifact def side(self): - def sources(started): + def sources(started, card_node=""): outputs = {} for seq, etype, nid, data in self.events: if seq < started and etype == "node.finished" and data.get("artifact"): diff --git a/backend/tests/test_compare_api.py b/backend/tests/test_compare_api.py index 541f87b7..41fa1b8f 100644 --- a/backend/tests/test_compare_api.py +++ b/backend/tests/test_compare_api.py @@ -476,3 +476,79 @@ async def test_real_runs_split_data_and_parameter_changes(client, engine_up, ord body = await get(client, f"/api/runs/{third}/compare/base") assert (body["base_run_id"], body["reason"]) == (second, "same_workflow") json.dumps(body, allow_nan=False) + + +# -------------------------------------------------------------------------- +# 子工作流里的口径卡：还原输入按卡所在那一层找产出 +# -------------------------------------------------------------------------- + + +def sub_card_graph(): + """子工作流：取数 → 原样解析 → 口径卡（东区金额读单元格；区域数读解析出来的整份结果的行，要从解析节点当时的产出还原）。""" + return chain( + node("s", "input", "开始"), + node("fetch", "tool", "区域取数", tool=f"db_query__{E2E_SOURCE}", args={"sql": E2E_SQL.replace( + "{{ input.start }}", "2026-08-31").replace("{{ input.end }}", "2026-09-06")}), + node("parse", "transform", "原样解析", mode="json", template="{{ nodes.fetch }}", assign_to="rows"), + node("card", "metrics", "子口径卡", caliber="周报口径", caliber_version="v1", metrics=[ + {"id": "east", "name": "东区金额", "unit": "元", "decimals": 1, "format": "thousands", + "expression": "cell(nodes.fetch, 0, 'gmv')"}, + {"id": "regions", "name": "区域数", "unit": "个", "decimals": 0, "format": "plain", + "expression": "len(nodes.parse.rows)"}]), + node("o", "output", "出口", fields=[{"name": "指标", "value": "{{ nodes.card.text }}"}]), + ) + + +async def test_cards_inside_a_subworkflow_restore_inputs_from_their_own_layer(client, engine_up, orders): + """子工作流里的卡（这次运行里的 id「sub/card」）取值链写的是子图里的 id（nodes.parse）：对比还原容器输入时要找「sub/parse」 + 的产出，不是父图里的 parse。父图里另有同名的 fetch、parse（查的是另一条 SQL），按层找才不会取错。""" + from tests import ledger_samples as S + + path, source_id = orders + sub_id = await S.publish("区域子流程", graph=sub_card_graph(), level="draft") + workflow = await new_workflow("子流程周报") + + def graph(): + return chain( + node("start", "input", "开始"), + # 父图里同名的取数、解析节点：一行，和子图里那份三行的结果不一样 + node("fetch", "tool", "父图取数", tool=f"db_query__{E2E_SOURCE}", + args={"sql": "SELECT COUNT(*) AS n FROM orders"}), + node("parse", "transform", "父图解析", mode="json", template="{{ nodes.fetch }}"), + node("sub", "subgraph", "区域子流程", workflow_id=sub_id, workflow_version=1), + node("out", "output", "成果", fields=[{"name": "指标", "value": "{{ nodes.sub.output.指标 }}"}]), + ) + + async def finished() -> str: + run = await run_manager.start(graph=graph(), input_payload={}, workflow_id=workflow, workflow_name="子流程周报") + for _ in range(400): + await asyncio.sleep(0.05) + async with SessionLocal() as session: + row = await session.get(Run, run.id) + if row.status in ("succeeded", "failed") and row.manifest_seq is not None: + assert row.status == "succeeded", row.error + return run.id + raise AssertionError(f"运行 {run.id} 没有结束") + + first = await finished() + db = sqlite3.connect(path) + db.execute("UPDATE orders SET amount = 1500.5 WHERE day = '2026-09-01'") + db.commit() + db.close() + await engines.invalidate(source_id) + second = await finished() + + body = await get(client, f"/api/runs/{second}/compare?base={first}") + metrics = {m["metric_id"] if "metric_id" in m else m["id"]: m for m in body["metrics"]} + east, regions = metrics["east"], metrics["regions"] + assert (east["classification"], east["base"], east["target"]) == ("data_only", 1200.5, 1500.5) + # 区域数：从子图里那次取数的产出还原（三行），两次一样 + assert (regions["base"], regions["target"]) == (3, 3) + assert regions["classification"] == "unchanged", regions + problems = [i.get("problem") for i in regions.get("inputs") or [] if i.get("problem")] + assert not problems, problems + # 区域数的那份结果顺着取值链认到子图里那次取数（两次之间改了数，快照不同），不是父图里同名的 fetch（两次一样） + [cell] = east["inputs"] + [rows] = regions["inputs"] + assert (rows["base_artifact"], rows["target_artifact"]) == (cell["base_artifact"], cell["target_artifact"]) + assert rows["base_artifact"] != rows["target_artifact"]