From 467f9599fd3c8c8d57736b5b7cc4313817f2a6d7 Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Mon, 5 Oct 2026 08:55:57 +0800 Subject: [PATCH] =?UTF-8?q?=E8=BF=90=E8=A1=8C=E5=AF=B9=E6=AF=94=EF=BC=9A?= =?UTF-8?q?=E8=A1=A5=E5=90=88=E5=B9=B6=E6=9F=A5=E8=AF=A2=E3=80=81=E5=BE=AA?= =?UTF-8?q?=E7=8E=AF=E9=87=8C=E5=A4=9A=E6=AC=A1=E6=89=A7=E8=A1=8C=E7=9A=84?= =?UTF-8?q?=E9=85=8D=E5=AF=B9=E6=B5=8B=E8=AF=95=EF=BC=8C=E5=8F=A3=E5=BE=84?= =?UTF-8?q?=E5=8D=A1=E8=AF=BB=E4=B8=8D=E5=87=BA=E6=9D=A5=E6=97=B6=E7=9A=84?= =?UTF-8?q?=E8=AF=B4=E6=B3=95=E8=A1=A5=E4=B8=8A=E3=80=8C=E7=9A=84=E3=80=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 合并查询只记在节点台账里、SQL 另在 merge.end 里，快照文件不在了照样比得了 SQL；循环里同一个节点执行几次的按先后 一一配上，多出来的一次标「删除」。口径卡指标集读不出来、被改过时的说明改成「口径卡「X」的指标集与哈希不一致」。 --- backend/app/engine/compare.py | 5 ++-- backend/tests/test_compare_api.py | 1 - backend/tests/test_compare_engine.py | 42 ++++++++++++++++++++++++++++ 3 files changed, 45 insertions(+), 3 deletions(-) diff --git a/backend/app/engine/compare.py b/backend/app/engine/compare.py index da41a670..f30dd816 100644 --- a/backend/app/engine/compare.py +++ b/backend/app/engine/compare.py @@ -317,11 +317,12 @@ def _cards(side: Side) -> dict[str, _Card]: def _card_problem(load: Callable[[str], Any], artifact: str) -> str: + """指标集为什么用不了，接在「口径卡「X」」后面。""" try: content = load(artifact) except ValueError: - return "指标集与哈希不一致，疑似被修改" - return "无法读取指标集" if content is None else "指标集的内容不完整" + return "的指标集与哈希不一致，疑似被修改" + return "的指标集无法读取" if content is None else "的指标集内容不完整" def _plan_card(side: Side, card: _Card, queries: list[_Query]) -> None: diff --git a/backend/tests/test_compare_api.py b/backend/tests/test_compare_api.py index 7e92273a..cd2cf0d9 100644 --- a/backend/tests/test_compare_api.py +++ b/backend/tests/test_compare_api.py @@ -28,7 +28,6 @@ from app.db.models import Artifact, DataSource, Run, RunEvent, Workflow from app.engine.runner import run_manager, verify_manifest from app.main import app from tests.fixtures.compare_runs import ( - BASE_ROWS, EAST, NEXT_ROWS, Fake, diff --git a/backend/tests/test_compare_engine.py b/backend/tests/test_compare_engine.py index c80ca91e..519fcba3 100644 --- a/backend/tests/test_compare_engine.py +++ b/backend/tests/test_compare_engine.py @@ -16,6 +16,7 @@ import json from decimal import Decimal from app.engine import compare +from app.engine.merge_query import MERGE_SOURCE from tests.fixtures.compare_runs import ( BASE_ROWS, EAST, @@ -530,6 +531,47 @@ def test_an_agent_reconfigured_with_the_same_sql_is_a_template_change(): assert metrics["east"]["classification"] == "data_only" +def test_a_merge_query_is_compared_by_its_merge_sql(): + """合并查询只记在节点的台账里（没有 tool.end），SQL 在快照和 merge.end 里；快照文件不在了也比得了。""" + def make(run_id, sql, *, drop=False): + run = Fake(run_id) + run.tool("fetch", region_sql()) + snap = {"columns": ["region", "gmv"], "rows": [["东区", 1.0]], "truncated": False, "sql": sql, + "source": MERGE_SOURCE, "sources": ["shop"]} + artifact = run.put(snap) + run.add("node.started", "join", {"node_type": "merge"}) + run.add("merge.end", "join", {"sql": sql, "query_artifact": artifact, "rows": 1}) + run.add("node.finished", "join", {"duration_ms": 1, "attempt": 1, "evidence": [ + {"kind": "query", "node_id": "join", "exec": 1, "artifact": artifact, "source": MERGE_SOURCE, + "columns": ["region", "gmv"], "rows": 1, "truncated": False}]}) + if drop: + del run.store[artifact] + return run + + result = run_compare(make("r1", "SELECT region, gmv FROM a WHERE gmv > 10"), + make("r2", "SELECT region, gmv FROM a WHERE gmv > 20")) + join = result_queries(result, "join") + assert (join["ordinal"], join["structure"], join["literals"]) == (1, "params_only", [{"base": "10", "target": "20"}]) + result = run_compare(make("r1", "SELECT region, gmv FROM a WHERE gmv > 10", drop=True), + make("r2", "SELECT region, gmv FROM a WHERE gmv > 10 AND region <> '西区'")) + join = result_queries(result, "join") + assert join["structure"] == "changed" and join["base_sql"] == "SELECT region, gmv FROM a WHERE gmv > 10" + assert "无法读取" in join["problem"] + + +def test_a_node_run_in_a_loop_is_paired_execution_by_execution(): + def make(run_id, regions): + run = Fake(run_id) + for region in regions: + run.tool("fetch", f"SELECT SUM(amount) AS gmv FROM orders WHERE region = '{region}'", ["gmv"], [[1.0]], + types={"gmv": "number"}) + return run + + result = run_compare(make("r1", ["东区", "南区", "西区"]), make("r2", ["东区", "北区"])) + assert [(q["ordinal"], q["structure"]) for q in result["queries"]] == [(1, "same"), (2, "params_only"), (3, "removed")] + assert result["queries"][1]["literals"] == [{"base": "'南区'", "target": "'北区'"}] + + def test_unpaired_queries_are_added_or_removed(): base = week("r1") target = week("r2")