From 787b2304dcbe055156d138202a724d70d72cc935 Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Sat, 3 Oct 2026 10:40:47 +0800 Subject: [PATCH] =?UTF-8?q?=E6=8C=87=E6=A0=87=E7=BB=8F=E8=BF=87=E5=90=88?= =?UTF-8?q?=E5=B9=B6=E6=9F=A5=E8=AF=A2=E4=B9=9F=E8=AE=A4=E5=BE=97=E5=87=BA?= =?UTF-8?q?=E6=9D=A5=E6=BA=90=E6=9F=A5=E8=AF=A2=E7=9A=84=20SQL=20=E6=A3=80?= =?UTF-8?q?=E6=9F=A5=E9=97=AE=E9=A2=98=EF=BC=9A=E5=90=88=E5=B9=B6=E5=BF=AB?= =?UTF-8?q?=E7=85=A7=E6=B2=A1=E6=9C=89=E8=87=AA=E5=B7=B1=E7=9A=84=20checks?= =?UTF-8?q?=EF=BC=8C=E9=A1=BA=E7=9D=80=20inputs=20=E5=BE=80=E4=B8=8B?= =?UTF-8?q?=E6=89=BE=EF=BC=8C=E6=BA=90=E5=BA=93=E4=B8=80=E5=AF=B9=E5=A4=9A?= =?UTF-8?q?=E9=87=8D=E5=A4=8D=E8=AE=A1=E7=AE=97=E7=9A=84=E6=95=B0=E4=B8=8D?= =?UTF-8?q?=E5=86=8D=E7=BB=8F=E4=B8=80=E6=AC=A1=E5=90=88=E5=B9=B6=E5=B0=B1?= =?UTF-8?q?=E5=AE=8C=E6=95=B4=E5=87=BA=E5=85=B7?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/app/engine/nodes/metrics.py | 23 +++++++++-- backend/tests/test_merge_sqlcheck.py | 60 ++++++++++++++++++++++++++++ 2 files changed, 80 insertions(+), 3 deletions(-) create mode 100644 backend/tests/test_merge_sqlcheck.py diff --git a/backend/app/engine/nodes/metrics.py b/backend/app/engine/nodes/metrics.py index b13d9878..9e41cd4f 100644 --- a/backend/app/engine/nodes/metrics.py +++ b/backend/app/engine/nodes/metrics.py @@ -33,6 +33,7 @@ from app.engine.expressions import ( substitute, ) from app.engine.labels import field_label +from app.engine.merge_query import MERGE_SOURCE from app.engine.state import GraphState, template_context @@ -393,13 +394,29 @@ def _result_artifact(path: str, tctx: dict[str, Any]) -> str | None: return None -def _error_checks(artifact: str) -> list[dict[str, Any]]: +def _error_checks(artifact: str, _seen: set[str] | None = None) -> list[dict[str, Any]]: + """这份查询快照对照数据目录查出的 error 级问题。 + + 合并查询的快照没有自己的 checks（它在内存 SQLite 上执行，不对照数据目录），但记着 inputs：顺着往下找。 + 不然源库那条查询一对多关联后重复计算了，经过一次合并，指标就看不出来、照样完整出具。 + """ + seen = _seen if _seen is not None else set() + if artifact in seen: + return [] + seen.add(artifact) try: snapshot = artifact_store.load(artifact) except Exception: # noqa: BLE001 - 快照读不出来、被改过：这里不报，证据层另有说法 return [] - checks = snapshot.get("checks") if isinstance(snapshot, dict) else None - return [c for c in checks if isinstance(c, dict) and c.get("level") == "error"] if isinstance(checks, list) else [] + if not isinstance(snapshot, dict): + return [] + checks = snapshot.get("checks") + found = [c for c in checks if isinstance(c, dict) and c.get("level") == "error"] if isinstance(checks, list) else [] + inputs = snapshot.get("inputs") if snapshot.get("source") == MERGE_SOURCE else None + for entry in inputs if isinstance(inputs, list) else []: + if isinstance(entry, dict) and isinstance(entry.get("artifact"), str): + found += _error_checks(entry["artifact"], seen) + return found def _sql_check_reason(failed: list[tuple[str, list[dict[str, Any]]]]) -> str: diff --git a/backend/tests/test_merge_sqlcheck.py b/backend/tests/test_merge_sqlcheck.py new file mode 100644 index 00000000..1be800ea --- /dev/null +++ b/backend/tests/test_merge_sqlcheck.py @@ -0,0 +1,60 @@ +"""合并查询把输入查询的 SQL 检查结果传到指标上。 + +口径卡判「来源查询有没有 error 级问题」读的是来源快照上的 checks（data/sqlcheck.py 在数据源查询时写）。 +合并查询在内存 SQLite 上执行，不对照数据目录检查，它的快照没有 checks——于是源库那条查询明明一对多关联后 +重复计算了，经过一次合并，指标就看不出来了，照样完整出具。合并快照记着 inputs（每个输入的快照 id）， +顺着它往下找才对得上。 +""" +from __future__ import annotations + +import app.engine.nodes.metrics as metrics_mod +from app.engine.merge_query import MERGE_SOURCE +from app.engine.nodes.metrics import _error_checks + +FANOUT = {"code": "fanout_sum", "level": "error", "message": "「订单」关联「订单明细」是一对多，求和会重复计算"} +NOTICE = {"code": "missing_valid_filter", "level": "warning", "message": "没有过滤作废记录"} + + +def _query(rows, checks=None): + snap = {"columns": ["日期", "门店", "销售额"], "rows": rows, "row_count": len(rows), "truncated": False} + return {**snap, "checks": checks} if checks else snap + + +def _merged(inputs): + return {"columns": ["日期", "门店", "销售额"], "rows": [["2026-05-01", "S01", 10]], "row_count": 1, + "truncated": False, "sql": "SELECT * FROM s JOIN v USING (日期, 门店)", "source": MERGE_SOURCE, + "inputs": [{"alias": a, "node_id": f"q_{a}", "artifact": art, "rows": 1} for a, art in inputs]} + + +def _store(monkeypatch, snapshots: dict): + monkeypatch.setattr(metrics_mod.artifact_store, "load", lambda artifact: snapshots.get(artifact)) + + +def test_an_error_in_a_merge_input_reaches_the_metric(monkeypatch): + _store(monkeypatch, { + "sales": _query([["2026-05-01", "S01", 10]], [FANOUT, NOTICE]), + "visits": _query([["2026-05-01", "S01", 3]]), + "merged": _merged([("s", "sales"), ("v", "visits")]), + }) + assert [c["code"] for c in _error_checks("merged")] == ["fanout_sum"] + + +def test_clean_inputs_leave_the_merge_clean(monkeypatch): + _store(monkeypatch, {"sales": _query([["2026-05-01", "S01", 10]], [NOTICE]), + "visits": _query([["2026-05-01", "S01", 3]]), + "merged": _merged([("s", "sales"), ("v", "visits")])}) + assert _error_checks("merged") == [] + + +def test_merges_of_merges_are_followed_and_cycles_do_not_hang(monkeypatch): + _store(monkeypatch, { + "sales": _query([["2026-05-01", "S01", 10]], [FANOUT]), + "inner": _merged([("s", "sales")]), + "outer": _merged([("m", "inner"), ("self", "outer")]), # 自引用：工件内容寻址做不出来，读坏的库也不能卡死 + }) + assert [c["code"] for c in _error_checks("outer")] == ["fanout_sum"] + + +def test_a_plain_query_snapshot_is_read_as_before(monkeypatch): + _store(monkeypatch, {"q": _query([["2026-05-01", "S01", 10]], [FANOUT, NOTICE])}) + assert [c["code"] for c in _error_checks("q")] == ["fanout_sum"]