From aee237f457c6dcb450319d5474554811b1801d74 Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Mon, 19 Oct 2026 10:01:05 +0800 Subject: [PATCH] =?UTF-8?q?=E6=8C=89=E8=A1=8C=E5=8F=B7=E5=8F=96=E5=A4=9A?= =?UTF-8?q?=E8=A1=8C=E6=9F=A5=E8=AF=A2=E9=87=8C=E7=9A=84=E4=B8=80=E6=A0=BC?= =?UTF-8?q?=EF=BC=8C=E9=87=8D=E8=B7=91=E3=80=81=E6=A0=B8=E5=AF=B9=E6=97=B6?= =?UTF-8?q?=E8=A1=8C=E5=BA=8F=E4=B8=8D=E5=AE=9A=E7=9A=84=E7=85=A7=E5=AE=9E?= =?UTF-8?q?=E6=A0=87=E5=87=BA=E6=9D=A5=EF=BC=9Asql=5Fsemantics.ordered=20/?= =?UTF-8?q?=20row=5Fstable=20=E8=AE=A4=E8=BF=99=E4=B8=80=E8=A1=8C=E9=87=8D?= =?UTF-8?q?=E8=B7=91=E6=97=B6=E8=BF=98=E6=98=AF=E4=B8=8D=E6=98=AF=E5=90=8C?= =?UTF-8?q?=E4=B8=80=E8=A1=8C=EF=BC=88=E6=9C=80=E5=A4=96=E5=B1=82=E5=B8=A6?= =?UTF-8?q?=20ORDER=20BY=20=E7=9A=84=E6=8C=89=E4=BD=8D=E7=BD=AE=E8=AE=A4?= =?UTF-8?q?=EF=BC=8C=E4=B8=80=E5=AE=9A=E5=8F=AA=E5=87=BA=E4=B8=80=E8=A1=8C?= =?UTF-8?q?=E7=9A=84=E7=AC=AC=200=20=E8=A1=8C=E8=AE=A4=E5=BE=97=E4=BD=8F?= =?UTF-8?q?=EF=BC=8C=E5=88=AB=E7=9A=84=E8=A1=8C=E5=BA=8F=E4=B8=8D=E5=AE=9A?= =?UTF-8?q?=EF=BC=89=E3=80=82=E5=8F=A3=E5=BE=84=E8=B4=A6=E6=9C=AC=E7=9A=84?= =?UTF-8?q?=E6=AF=94=E5=AF=B9=E8=AE=B0=E4=B8=8B=20ordered=E3=80=81one=5Fro?= =?UTF-8?q?w=EF=BC=8C=E6=8C=87=E6=A0=87=E6=A0=87=E7=AD=BE=E6=8C=89?= =?UTF-8?q?=E3=80=8C=E5=90=8C=E4=B8=80=E8=A1=8C=E3=80=8D=E8=AE=A4=E6=9D=A5?= =?UTF-8?q?=E8=87=AA=E5=8F=A3=E5=BE=84=E5=8F=AA=E5=9C=A8=E8=A1=8C=E8=AE=A4?= =?UTF-8?q?=E5=BE=97=E4=BD=8F=E6=97=B6=E6=88=90=E7=AB=8B=EF=BC=8C=E8=A1=8C?= =?UTF-8?q?=E5=BA=8F=E4=B8=8D=E5=AE=9A=E7=9A=84=E6=A0=87=E7=9B=B8=E5=85=B3?= =?UTF-8?q?=E3=80=81=E5=86=99=E6=98=8E=E5=8E=9F=E5=9B=A0=EF=BC=9B=E5=8F=A3?= =?UTF-8?q?=E5=BE=84=E5=8F=AC=E5=9B=9E=E9=87=8C=E4=B8=A4=E7=89=88=E9=83=BD?= =?UTF-8?q?=E6=8C=89=E8=A1=8C=E5=8F=B7=E5=8F=96=E4=B8=80=E6=A0=BC=E3=80=81?= =?UTF-8?q?=E6=9F=A5=E8=AF=A2=E5=8F=88=E6=B2=A1=E6=9C=89=20ORDER=20BY=20?= =?UTF-8?q?=E7=9A=84=E6=8C=87=E6=A0=87=E8=AE=B0=E8=A1=8C=E5=BA=8F=E4=B8=8D?= =?UTF-8?q?=E5=AE=9A=EF=BC=88changed=20=E4=B8=BA=E7=A9=BA=EF=BC=89?= =?UTF-8?q?=EF=BC=8C=E5=8F=AC=E5=9B=9E=E5=BE=85=E5=8A=9E=E4=B8=8A=E5=86=99?= =?UTF-8?q?=E3=80=8C=E8=A1=8C=E5=BA=8F=E4=B8=8D=E5=AE=9A=E3=80=8D=EF=BC=8C?= =?UTF-8?q?=E4=B8=8D=E6=8C=89=E4=BD=8D=E7=BD=AE=E7=A1=AC=E6=AF=94=EF=BC=9B?= =?UTF-8?q?=E5=AF=B9=E8=B4=A6=E7=9A=84=E5=A4=9A=E8=A1=8C=E5=AF=B9=E9=BD=90?= =?UTF-8?q?=E6=AF=8F=E4=B8=80=E7=BB=84=E5=85=88=E6=A0=B8=E5=AF=B9=E9=82=A3?= =?UTF-8?q?=E4=B8=80=E8=A1=8C=E7=9A=84=E7=BB=B4=E5=BA=A6=E5=B0=B1=E6=98=AF?= =?UTF-8?q?=E5=AE=83=EF=BC=8C=E9=87=8D=E8=B7=91=E6=97=B6=E7=BC=BA=E4=BA=86?= =?UTF-8?q?=E4=B8=80=E8=A1=8C=E3=80=81=E5=A4=9A=E5=87=BA=E4=B8=80=E8=A1=8C?= =?UTF-8?q?=E9=94=99=E4=BD=8D=E7=9A=84=E8=AE=B0=E7=A9=BA=E5=80=BC=EF=BC=8C?= =?UTF-8?q?=E4=B8=8D=E6=8B=BF=E5=88=AB=E7=9A=84=E7=BB=B4=E5=BA=A6=E7=9A=84?= =?UTF-8?q?=E6=95=B0=E5=86=92=E5=85=85?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/app/data/sql_semantics.py | 24 +++++ backend/app/ledger/labels.py | 16 ++++ backend/app/ledger/match.py | 8 +- backend/app/ledger/recall.py | 47 +++++++++- backend/app/ledger/reconcile.py | 28 +++++- backend/tests/test_ledger_projection.py | 5 +- backend/tests/test_ledger_row_order.py | 119 ++++++++++++++++++++++++ frontend/src/lib/termsRecall.ts | 2 + frontend/src/run/RecallTask.tsx | 13 ++- frontend/src/typesRecall.ts | 5 +- 10 files changed, 250 insertions(+), 17 deletions(-) create mode 100644 backend/tests/test_ledger_row_order.py diff --git a/backend/app/data/sql_semantics.py b/backend/app/data/sql_semantics.py index 0d1f81a7..dd3ae597 100644 --- a/backend/app/data/sql_semantics.py +++ b/backend/app/data/sql_semantics.py @@ -9387,6 +9387,30 @@ def one_row(sql: str, dialect: str | None = None) -> bool: return _aggregating(root) +def ordered(sql: str, dialect: str | None = None) -> bool: + """这条查询最外层带 ORDER BY（一段 SELECT 或 UNION 整体的）：结果的行序由查询定，按行号取的那一格重跑时还是那一行（排序键 + 相同的几行之间的先后另说，按位置认）。认不出的为假。""" + from sqlglot import exp + + read = _read(dialect) + try: + root, _ = _parse(_mysql_dashes(sql) if read == "mysql" and isinstance(sql, str) else sql, read) + except Exception: # noqa: BLE001 - 解析不了就是认不出 + return False + if not isinstance(root, (exp.Select, exp.Union)): + return False + order = root.args.get("order") + return order is not None and bool(order.expressions) + + +def row_stable(sql: str, row: int, dialect: str | None = None) -> bool: + """按行号取这条查询的第 row 行，重跑、核对时还是不是同一行：最外层带 ORDER BY 的按位置认；一定只出一行的（one_row）第 0 行 + 认得住；别的（多行、又没有 ORDER BY）行序不定——库可以按任何顺序交回这些行，第 2 行重跑时可能是另一行。""" + if ordered(sql, dialect): + return True + return row == 0 and one_row(sql, dialect) + + def output_names(sql: str, dialect: str | None = None, *, template: bool = False) -> list[str] | None: """最外层选择列表各项的输出名（按位置；同 _result_columns，不展开 *：有 * 的为 None）。template 为真时 sql 是口径 模板，按哨兵渲染之后认。解析不了、不是一段 SELECT 的为 None。""" diff --git a/backend/app/ledger/labels.py b/backend/app/ledger/labels.py index e876f6c2..f53861e4 100644 --- a/backend/app/ledger/labels.py +++ b/backend/app/ledger/labels.py @@ -47,6 +47,9 @@ METRIC_GONE = "指标集里没有这个指标，没有比对" UNFINISHED = "比对没有做完" #: related 的几种说明 FORMULA_DIFFERS = "取数和口径等价，但指标的算法和口径卡上的不同" +#: 指标按「同一行」认来自口径，可那一行重跑时不一定还是它：多行结果、查询又没有 ORDER BY（sql_semantics.row_stable） +ROW_ORDER_UNSTABLE = ("指标取的是多行结果里按行号的一格，查询没有 ORDER BY，行序不定：重新执行时这一行可能是另一行，" + "不能认定对应口径的同一格") INPUTS_ELSEWHERE = "指标有的输入不来自查询（运行输入、整形或代码），不能算来自口径" MIXED_CALIBERS = "指标的几个输入分别对上了不同的口径" BINDING_CONFLICT = "指标的几个输入对上了同一份口径，但参数不一致" @@ -143,6 +146,15 @@ def cell_label(artifact: str, column: Any, queries: Mapping[str, Mapping[str, An return {**_unknown(str(query.get("reason") or UNFINISHED)), "query": artifact} +def _row_stable(query: Mapping[str, Any], item: Mapping[str, Any]) -> bool: + """这个输入取的那一行重跑时还认得住（match 记的 ordered / one_row，sql_semantics.row_stable 的认法）：最外层带 ORDER BY， + 或者一定只出一行、取的是第 0 行。老的比对记录没记 ordered 的照旧（当时不判）。""" + if "ordered" not in query: + return True + where = item.get("locator") if isinstance(item.get("locator"), dict) else {} + return bool(query.get("ordered")) or (where.get("row") in (0, None) and bool(query.get("one_row"))) + + def _leaf_token(node_id: Any, row: Any, column: Any) -> str: """把指标式里的一个输入换成「口径的哪个取数节点、第几行、哪一列」：两边的式子按这个换完再逐字比。""" return f"⟦{node_id}|{row}|{column}⟧" @@ -307,6 +319,10 @@ def metric_label(card: Mapping[str, Any] | None, metric_id: Any, queries: Mappin return {"kind": "related", "related": nearby, "reason": MIXED_CALIBERS, **queried} related = next((q for q in queries_found if q.get("status") == "related"), None) return {**_related(related), **queried} if related is not None else {"kind": "reasoning", **queried} + if unstable := [i for i, q in zip(from_queries, queries_found) if not _row_stable(q, i)]: + # 按「同一行」认来自口径（_leaf_token 记的是行号）：多行结果又没有 ORDER BY 的，重跑时第几行不一定是哪一行， + # 照实标相关、写明行序不定，不认成来自口径 + return {"kind": "related", "related": nearby, "reason": ROW_ORDER_UNSTABLE, **queried} reason = FORMULA_DIFFERS for key in sorted(common): chosen = [next(c for c in _proven(q) if c.get("caliber") == key) for q in queries_found] diff --git a/backend/app/ledger/match.py b/backend/app/ledger/match.py index f5d72135..3a4b12c4 100644 --- a/backend/app/ledger/match.py +++ b/backend/app/ledger/match.py @@ -98,6 +98,8 @@ SNAPSHOT_TAMPERED = "查询快照与哈希不一致，疑似被修改，不比 SNAPSHOT_MISSING = "查询快照读不出来，不比对" NO_SQL = "查询快照里没有 SQL，不比对" PARSE_FAILED = "这次的 SQL 无法解析，没有比对" +#: 指标按「同一行」认来自口径，可那一行重跑时不一定还是它（说法在 labels，这里给调用方一个名字） +ROW_ORDER_UNSTABLE = ledger_labels.ROW_ORDER_UNSTABLE MERGED = "合并查询的格不和口径比对（它的数来自各个输入的查询）" COMPOSED_UNPROVEN = "这条查询出自助手选用的口径，但和口径的取数没能证明一致（可能被改过）" MORE_CANDIDATES = "另有 {n} 份相关口径没有比对（每条查询最多比 {max} 份）" @@ -546,7 +548,7 @@ def compare_queries(queries: list[Query], calibers: list[Caliber], *, deadline: 等价）/ none（没有相关的口径）/ skipped（没比完：超时、解析失败、快照读不出）。composed 是 composed_nodes 的结果， 不给就没有助手选用的。referenced 是报告引用了哪条查询的哪几列（labels.referenced_columns：{查询快照: {列名}}）： 给了的，整条比不等价时按引用的列投影再比（_projected）；不给就只整条比。纯计算，从不抛异常。评测逐题比 SQL 也调它。""" - from app.data.sql_semantics import match_template, normalize_sql, one_row + from app.data.sql_semantics import match_template, normalize_sql, one_row, ordered out: list[dict[str, Any]] = [] for query in queries: @@ -567,6 +569,9 @@ def compare_queries(queries: list[Query], calibers: list[Caliber], *, deadline: if normalize_sql(query.sql, query.dialect).parse_error: out.append({**row, "status": "skipped", "reason": PARSE_FAILED, "candidates": []}) continue + # 按行号取的那一格重跑时还是不是同一行（sql_semantics.row_stable）：指标标签按「同一行」认来自口径时要看它 + row["ordered"] = ordered(query.sql, query.dialect) + row["one_row"] = one_row(query.sql, query.dialect) candidates: list[dict[str, Any]] = [] for caliber, fetch in pool[:MAX_CANDIDATES]: if deadline is not None and time.monotonic() > deadline: @@ -600,7 +605,6 @@ def compare_queries(queries: list[Query], calibers: list[Caliber], *, deadline: if wanted is not None and len(wanted) >= 2 and status == "related": # 整条（按全部引用列投影）没证明：被引用的几列逐格各比一次，每个数按它自己那一格的结论标（labels.cell_label） row["cells"] = _cells(query, wanted, pool[:MAX_CANDIDATES], candidates, chosen, deadline) - row["one_row"] = one_row(query.sql, query.dialect) out.append(row) return out diff --git a/backend/app/ledger/recall.py b/backend/app/ledger/recall.py index 62bf2dbe..bcb9d7b8 100644 --- a/backend/app/ledger/recall.py +++ b/backend/app/ledger/recall.py @@ -364,15 +364,48 @@ def recall_card(pairs: Sequence[Pair]) -> tuple[dict[str, Any], list[dict[str, A _metric(f"new_{k}", f"{name} · 新版 v{pair.new.version}", new_expr, nm), _metric(f"diff_{k}", f"{name} · 两版之差", f"({new_expr}) - ({old_expr})", nm)] # 旧版的式子（定义，不是数）一并记下：问数回答当时的值要按它找到答案里对应的那一格（original_values） - meta.append({"k": k, "card": pair.old.node_id, "metric": m["id"], "name": name, - "unit": str(nm.get("unit") or ""), "decimals": nm.get("decimals"), "format": nm.get("format"), - "expression": str(m.get("expression") or "")}) + item = {"k": k, "card": pair.old.node_id, "metric": m["id"], "name": name, + "unit": str(nm.get("unit") or ""), "decimals": nm.get("decimals"), "format": nm.get("format"), + "expression": str(m.get("expression") or "")} + if not (_rows_stable(pair.old, m) and _rows_stable(pair.new, nm)): + # 两版都按行号取多行结果里的一格、查询又没有 ORDER BY：两次查询交回的行序可能不同，第几行不一定是同一行， + # 按位置比出来的「变了 / 没变」都不可信（compare 照实记行序不定） + item["row_order"] = "unstable" + meta.append(item) k += 1 node = derive._node(derive.RECALL_CARD, "metrics", "召回卡", caliber="口径召回", caliber_version="v1", on_missing="null", metrics=metrics) return node, meta +def _rows_stable(pinned: Any, metric: Mapping[str, Any]) -> bool: + """这个指标按行号取的每一格（cell(nodes.<取数>, 行, '列')），在那一版的取数 SQL 上重跑时还是同一行（sql_semantics.row_stable）。 + 行号、SQL 认不出来的照旧当认得住（以前就是按位置比的，不凭猜测改结论）。""" + import ast + + from app.data.sql_semantics import row_stable + from app.engine.expressions import ExpressionError, cell_parts, leaf_refs, parse_expression + + sql_of = {str(n.get("id")): ((n.get("data") or {}).get("config") or {}).get("args", {}).get("sql") + for n in pinned.fetches if isinstance(n, dict)} + try: + leaves = leaf_refs(parse_expression(str(metric.get("expression") or ""))[0]) + except ExpressionError: + return True + for leaf in leaves: + parts = cell_parts(str(leaf)) + if parts is None or not parts[0].startswith("nodes."): + continue + sql = sql_of.get(parts[0].removeprefix("nodes.")) + try: + row = ast.literal_eval(parts[1].strip()) + except (ValueError, SyntaxError): + continue + if isinstance(sql, str) and isinstance(row, int) and not row_stable(sql, row): + return False + return True + + def build_graph(group: Group, pairs: Sequence[Pair]) -> tuple[Any, list[dict[str, Any]]]: """一组的召回图（derive.recall_graph）和比较说明。同样的输入永远得到同一张图。""" from app.ledger import derive @@ -607,11 +640,14 @@ def compare(meta: Sequence[Mapping[str, Any]], values: Mapping[str, Mapping[str, for item in meta: k = item.get("k") old, new, diff = (values.get(f"{p}_{k}") or {} for p in ("old", "new", "diff")) - changed = str(old.get("rendered")) != str(new.get("rendered")) + unstable = item.get("row_order") == "unstable" + # 行序不定的（recall_card 记的）：按位置比不出变没变，changed 为空，照实记 row_order，不假装对上 + changed = None if unstable else str(old.get("rendered")) != str(new.get("rendered")) out.append({"k": k, "card": item.get("card"), "metric": item.get("metric"), "name": item.get("name"), "unit": item.get("unit"), "old": old.get("value"), "new": new.get("value"), "diff": diff.get("value"), "old_text": old.get("rendered"), "new_text": new.get("rendered"), - "diff_text": diff.get("rendered"), "changed": changed}) + "diff_text": diff.get("rendered"), "changed": changed, + **({"row_order": "unstable"} if unstable else {})}) return out @@ -810,6 +846,7 @@ async def task_payload(session: Any, task: Any, principal: Any) -> dict[str, Any cells[side] = {"value": None if masked else value, "text": "已遮罩" if masked else text, "metric": f"{side}_{item['k']}", "masked": masked} metrics.append({"k": item["k"], "name": item["name"], "unit": item["unit"], "changed": item["changed"], + **({"row_order": item["row_order"]} if item.get("row_order") else {}), "card": item["card"], "metric": item["metric"], **cells}) answers = [] for use in _uses_of(origin, _owner_of_key(key)): diff --git a/backend/app/ledger/reconcile.py b/backend/app/ledger/reconcile.py index db2edfff..6ac79ec5 100644 --- a/backend/app/ledger/reconcile.py +++ b/backend/app/ledger/reconcile.py @@ -296,15 +296,37 @@ def tail_nodes(plan: Plan, *, card_id: str, fetch_id: str) -> list[dict[str, Any sql=align_sql(plan.keys, plan.pairs, plan.key_values))) for i, row in enumerate(plan.key_values): where = "，".join(f"{ours}={v}" for (ours, _), v in zip(plan.keys, row)) + # 先核对第 i 行就是这一组的维度：重跑时这次的结果缺了一行、某个维度多出一行（口径那边重复），合并结果会错位， + # 按位置取到的是别的维度的数。对不上的记空值（对账卡 on_missing=null），不拿别的维度的数冒充 + guard = _row_guard(i, plan.keys, row) for j, pair in enumerate(plan.pairs): - metrics += _four(f"{i}_{j}", f"{where} · {pair.name or pair.ours}", - f"cell(nodes.{ALIGN_ID}, {i}, 'ours_{j}')", f"cell(nodes.{ALIGN_ID}, {i}, 'cal_{j}')", - pair) + ours = f"cell(nodes.{ALIGN_ID}, {i}, 'ours_{j}')" + theirs = f"cell(nodes.{ALIGN_ID}, {i}, 'cal_{j}')" + if guard: + ours, theirs = f"({ours} if {guard} else None)", f"({theirs} if {guard} else None)" + metrics += _four(f"{i}_{j}", f"{where} · {pair.name or pair.ours}", ours, theirs, pair) nodes.append(_node(RECONCILE_CARD, "metrics", "对账卡", caliber="口径对账", caliber_version="v1", on_missing="null", metrics=metrics)) return nodes +def _row_guard(i: int, keys: list[tuple[str, str]], values: tuple[Any, ...]) -> str: + """对齐结果第 i 行是不是这一组（维度列的值都对得上）的条件式；维度值是空值的那一列不核对（表达式里没有空值的写法）。""" + import ast + + from app.engine.expressions import _literal + + parts = [f"cell(nodes.{_align_id()}, {i}, {ours!r}) == {ast.unparse(_literal(v))}" + for (ours, _), v in zip(keys, values) if v is not None and not isinstance(v, (list, dict))] + return " and ".join(parts) + + +def _align_id() -> str: + from app.ledger.derive import ALIGN_ID + + return ALIGN_ID + + def _numeric(rows: list[Any], index: int) -> bool: values = [r[index] for r in rows if isinstance(r, (list, tuple)) and index < len(r) and r[index] is not None] return bool(values) and all(isinstance(v, (int, float)) and not isinstance(v, bool) for v in values) diff --git a/backend/tests/test_ledger_projection.py b/backend/tests/test_ledger_projection.py index 9fa3a257..6ab642fd 100644 --- a/backend/tests/test_ledger_projection.py +++ b/backend/tests/test_ledger_projection.py @@ -131,8 +131,9 @@ def test_without_a_projection_every_pair_is_exactly_the_old_result(): ] for sql, columns, cited in cases: row = compare(sql, columns, cited) - # 逐格比（algo 4）是另加的 cells、one_row 两个键：整条查询这一层的结论和候选照旧一字不差 - assert {k: v for k, v in row.items() if k not in ("cells", "one_row")} == compare(sql, columns, None), sql + # 逐格比（algo 4）是另加的 cells 键（one_row、ordered 两边都记：行认不认得住，指标标签要看）：整条查询这一层的 + # 结论和候选照旧一字不差 + assert {k: v for k, v in row.items() if k != "cells"} == compare(sql, columns, None), sql assert ("cells" in row) == (len(cited) >= 2 and row["status"] == "related"), sql diff --git a/backend/tests/test_ledger_row_order.py b/backend/tests/test_ledger_row_order.py new file mode 100644 index 00000000..17bad01d --- /dev/null +++ b/backend/tests/test_ledger_row_order.py @@ -0,0 +1,119 @@ +"""按行号取多行查询里的一格，重跑、核对时行序不一定一样（next-round 验收遗留）： + +- 认法（data/sql_semantics.row_stable）：最外层带 ORDER BY 的按位置认；一定只出一行的第 0 行认得住；别的（多行、又没有 + ORDER BY）行序不定——库可以按任何顺序交回这些行； +- 口径账本的指标标签：答案的一格和口径卡的一格按「同一行」认来自口径，只在行认得住时成立；行序不定的照实标「相关」、 + 写明行序不定，不再标成来自口径； +- 口径召回：两版都按行号取一格、查询又行序不定的指标，照实记「行序不定」（changed 为空），不按位置硬比； +- 对账：多行结果按维度对齐之后，每一组对账指标先核对那一行的维度是不是它（重跑时缺了一行、多了一行，后面的行会错位）， + 对不上的记空值，不拿别的维度的数冒充。 +数据是虚构店铺库。 +""" +from __future__ import annotations + +import pytest + +from app.data.sql_semantics import ordered, row_stable +from app.engine.runner import run_manager +from app.ledger import match as ledger_match +from app.ledger import recall as ledger_recall +from app.ledger import reconcile as ledger_reconcile +from tests import ledger_samples as S + +UNORDERED = ("SELECT region AS 区域, SUM(amount) AS gmv FROM orders WHERE status = 'paid' " + "AND paid_at >= '{{ input.start }}' AND paid_at < '{{ input.end }}' GROUP BY region") +UNORDERED_ALL = ("SELECT region AS 区域, SUM(amount) AS gmv FROM orders WHERE status = 'paid' " + "AND paid_at >= '2026-09-01' AND paid_at < '2026-09-08' GROUP BY region") + + +def test_which_rows_survive_a_rerun(): + assert ordered("SELECT a, b FROM t ORDER BY a") and not ordered("SELECT a, b FROM t") + assert row_stable("SELECT a, SUM(b) FROM t GROUP BY a ORDER BY a", 3) + assert row_stable("SELECT SUM(b) FROM t", 0) and not row_stable("SELECT SUM(b) FROM t", 1) + assert not row_stable("SELECT a, SUM(b) FROM t GROUP BY a", 0) + assert not row_stable("这不是 SQL", 0) + + +@pytest.fixture +async def shop(tmp_path): + await run_manager.setup() + await S.fresh_source(tmp_path / "shop.db") + yield + await ledger_match.drain() + await run_manager.shutdown() + + +async def _label(monkeypatch, caliber_sql: str, answer_sql: str, row: int) -> dict: + from tests.test_ledger_match import matched, only_label + + await S.publish(graph=S.caliber_graph(caliber_sql, expression=f"cell(nodes.fetch, {row}, 'gmv')")) + run = await S.explore(monkeypatch, answer_sql, text="这一区销售额 [[m:gmv]]。", + card_expression=f"cell(nodes.q, {row}, 'gmv')") + recorded, _ = await matched(run.id) + return only_label(recorded.content) + + +async def test_an_unordered_multi_row_cell_is_not_claimed_to_come_from_the_caliber(shop, monkeypatch): + label = await _label(monkeypatch, UNORDERED, UNORDERED_ALL, 1) + assert label["kind"] == "related" and label["reason"] == ledger_match.ROW_ORDER_UNSTABLE, label + + +async def test_with_order_by_the_same_row_still_comes_from_the_caliber(shop, monkeypatch): + label = await _label(monkeypatch, UNORDERED + " ORDER BY region", UNORDERED_ALL + " ORDER BY region", 1) + assert label["kind"] == "caliber", label + + +def _pinned(sql: str, expression: str, version: int): + from app.ledger.derive import Pinned + + graph = S.caliber_graph(sql, expression=expression) + card = next(n for n in graph["nodes"] if n["id"] == "card") + fetch = next(n for n in graph["nodes"] if n["id"] == "fetch") + return Pinned(workflow_id="wf", workflow_name="周销售", version=version, node_id="card", level="governed", + published=True, retired=False, card=card, fetches=(fetch,), reads=(("fetch", S.tool(), ("gmv",)),), + entry_fields=(), defaults={}) + + +def test_recall_marks_unordered_row_comparisons_instead_of_comparing_by_position(): + pair = ledger_recall.Pair(old=_pinned(UNORDERED, "cell(nodes.fetch, 1, 'gmv')", 1), + new=_pinned(UNORDERED + " AND 1 = 1", "cell(nodes.fetch, 1, 'gmv')", 2)) + _, meta = ledger_recall.recall_card([pair]) + assert meta[0]["row_order"] == "unstable" + values = {"old_0": {"value": 1.0, "rendered": "1"}, "new_0": {"value": 2.0, "rendered": "2"}, + "diff_0": {"value": 1.0, "rendered": "1"}} + [item] = ledger_recall.compare(meta, values) + assert item["changed"] is None and item["row_order"] == "unstable" + # 带 ORDER BY 的照常比 + pair = ledger_recall.Pair(old=_pinned(UNORDERED + " ORDER BY region", "cell(nodes.fetch, 1, 'gmv')", 1), + new=_pinned(UNORDERED + " ORDER BY region", "cell(nodes.fetch, 1, 'gmv')", 2)) + _, meta = ledger_recall.recall_card([pair]) + assert "row_order" not in meta[0] + [item] = ledger_recall.compare(meta, values) + assert item["changed"] is True + + +def test_reconcile_checks_each_aligned_row_is_the_one_it_stands_for(): + plan = ledger_reconcile.Plan(pairs=[ledger_reconcile.Pair(ours="gmv", column="gmv")], rows=[0, 1], + keys=[("区域", "区域")], key_values=[("east",), ("west",)]) + nodes = ledger_reconcile.tail_nodes(plan, card_id="card", fetch_id="fetch") + card = next(n for n in nodes if n["id"] == "reconcile") + metrics = {m["id"]: m["expression"] for m in card["data"]["config"]["metrics"]} + from app.engine.expressions import eval_expression + + def run(rows): + table = {"columns": ["区域", "ours_0", "cal_0"], "rows": rows} + return {mid: _eval(eval_expression, expr, table) for mid, expr in metrics.items() if mid.startswith("ours_")} + + # 重跑时东区那一行没了：西区挪到了第 0 行。以前东区那一组拿的是西区的数 + got = run([["west", 58.0, 20.0]]) + assert got["ours_0_0"] is None, "东区对不上：记空值，不拿西区的数冒充" + assert run([["east", 163.0, 150.5], ["west", 58.0, 20.0]]) == {"ours_0_0": 163.0, "ours_1_0": 58.0} + + +def _eval(fn, expression, table): + from app.engine.expressions import CellError, ExpressionError + + try: + return fn(expression, {"nodes": {"align": table}}) + except (CellError, ExpressionError, IndexError, KeyError, TypeError): + return None diff --git a/frontend/src/lib/termsRecall.ts b/frontend/src/lib/termsRecall.ts index 0cc978be..a1931a94 100644 --- a/frontend/src/lib/termsRecall.ts +++ b/frontend/src/lib/termsRecall.ts @@ -99,6 +99,8 @@ export const RECALL_TEXT = { original: '原答案当时的值', changed: '变了', same: '没变', + rowOrder: '行序不定', + rowOrderHint: '取的是多行结果里按行号的一格，查询没有排序，重新执行时这一行可能是另一行，两版按位置比不出变没变', masked: '已遮罩', missing: '—', answers: '用到这份口径的答案', diff --git a/frontend/src/run/RecallTask.tsx b/frontend/src/run/RecallTask.tsx index f73b8c25..145e89a7 100644 --- a/frontend/src/run/RecallTask.tsx +++ b/frontend/src/run/RecallTask.tsx @@ -121,12 +121,17 @@ function Metrics({ payload, onOpen }: { payload: Partial; onO {metrics.map((m) => ( - + {m.name} - - {m.changed ? T.changed : T.same} - + {m.row_order === 'unstable' ? ( + {T.rowOrder} + ) : ( + + {m.changed ? T.changed : T.same} + + )} {(['old', 'new', 'diff'] as const).map((side, i) => ( diff --git a/frontend/src/typesRecall.ts b/frontend/src/typesRecall.ts index 1f478a4a..90aca8c8 100644 --- a/frontend/src/typesRecall.ts +++ b/frontend/src/typesRecall.ts @@ -151,7 +151,10 @@ export interface RecallTaskPayload { k: number name: string unit: string - changed: boolean + /** 行序不定（row_order）时为 null：按位置比不出变没变 */ + changed: boolean | null + /** 两版都按行号取多行结果里的一格、查询没有 ORDER BY：重跑时第几行不一定是同一行（服务端 ledger/recall.recall_card） */ + row_order?: 'unstable' card: string metric: string old: RecallCell