mirror of
https://github.com/ThearchyHelios/AgentLab.git
synced 2026-11-15 06:31:58 +00:00
指标经过合并查询也认得出来源查询的 SQL 检查问题:合并快照没有自己的 checks,顺着 inputs 往下找,源库一对多重复计算的数不再经一次合并就完整出具
This commit is contained in:
@@ -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:
|
||||
|
||||
@@ -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"]
|
||||
Reference in New Issue
Block a user