合并查询的警告带上结构化字段:键类型不一致带 keys、detail,行数放大带 rows、input、input_rows、duplicates,结果截断带 rows、max_rows,随运行日志一起发出,界面读字段,不再从中文原话里解析

This commit is contained in:
ThearchyHelios
2026-10-04 02:17:32 +08:00
parent 2c160f0ac9
commit 9c8bed2f1b
3 changed files with 14 additions and 4 deletions
+5 -3
View File
@@ -834,7 +834,8 @@ def _warnings(stmt: str, inputs: list[MergeInput], tables: dict[str, list[str]],
else:
tail = ("SQLite 比较时会做隐式转换:文本形式的编号(如 '001'、'01')都会等于数值 1,也可能完全匹配不上。"
"请在源查询中统一类型,或在合并 SQL 中用 CAST 明确转换")
out.append({"code": "key_type_mismatch", "keys": [f"{a}.{ac}", f"{b}.{bc}"],
# keys、detail 是给界面读的结构化字段(运行时间线的标题、副标题),不让它从 message 里解析中文
out.append({"code": "key_type_mismatch", "keys": [f"{a}.{ac}", f"{b}.{bc}"], "detail": f"{left},{right}",
"message": f"合并键类型不一致:{left},{right}。{tail}"})
largest = max(inputs, key=lambda inp: len(inp.snapshot["rows"]))
biggest = len(largest.snapshot["rows"])
@@ -846,12 +847,13 @@ def _warnings(stmt: str, inputs: list[MergeInput], tables: dict[str, list[str]],
keys[alias].append(col)
dups = [_duplicates(alias, snaps[alias], cols) for alias, cols in keys.items()]
detail = ";".join(d for d in dups if d)
out.append({"code": "rows_grew", "message": (
out.append({"code": "rows_grew", "rows": len(rows), "input": largest.alias, "input_rows": biggest,
**({"duplicates": detail} if detail else {}), "message": (
f"合并结果有 {len(rows)} 行,多于行数最多的输入「{largest.alias}」({biggest} 行):合并键可能不唯一,"
"同一行被重复匹配。" + (f"{detail}。" if detail else "")
+ "请检查合并条件是否覆盖了全部键(例如同时按日期和门店),或先在源库里聚合到相同粒度")})
if truncated:
out.append({"code": "result_truncated", "message": (
out.append({"code": "result_truncated", "rows": len(rows), "max_rows": limits.max_rows, "message": (
f"合并结果超过上限({limits.max_rows} 行或约 {max(1, limits.max_bytes // 1_000_000)} MB),只保留了前 "
f"{len(rows)} 行,下游拿到的不是完整结果。请在合并 SQL 中聚合或加条件缩小范围")})
return out
+3 -1
View File
@@ -108,8 +108,10 @@ async def run_merge(state: GraphState, ctx: NodeContext) -> dict[str, Any]:
preview_rows=outcome.rows[:PREVIEW_ROWS], duration_ms=elapsed, query_artifact=artifact,
lineage=outcome.lineage is not None)
for warning in outcome.warnings:
# 结构化字段(keys、detail、rows、input、input_rows、duplicates、max_rows)随日志带上:界面读字段,不解析中文
fields = {k: v for k, v in warning.items() if k not in ("code", "message")}
ctx.emit(EventType.LOG, level="warn", message=warning["message"],
code=LOG_CODES.get(warning["code"], "merge_warning"))
code=LOG_CODES.get(warning["code"], "merge_warning"), **fields)
output: dict[str, Any] = {
"artifact": artifact, "columns": outcome.columns, "rows": outcome.rows, "row_count": len(outcome.rows),
+6
View File
@@ -462,6 +462,12 @@ async def test_warnings_reach_the_output_and_the_events(monkeypatch, sources):
logs = [e for e in await events(row.id, "log", "merge") if e.get("level") == "warn"]
assert sorted(e["code"] for e in logs) == ["merge_key_type", "merge_rows_grew"]
assert all(e["message"] for e in logs)
# 结构化字段随日志带上:运行时间线读字段,不从中文里解析
key = next(e for e in logs if e["code"] == "merge_key_type")
assert key["keys"] == ["s.门店", "v.门店"] and key["detail"].startswith("s.门店 是文本") and "v.门店 是数值" in key["detail"]
grew = next(e for e in logs if e["code"] == "merge_rows_grew")
assert grew["rows"] == done["rows"] and grew["input"] in ("s", "v") and isinstance(grew["input_rows"], int)
assert grew["rows"] > grew["input_rows"] and "有重复的键" in grew["duplicates"]
async def test_rerunning_the_node_on_the_same_inputs_gives_the_same_snapshot(monkeypatch, sources):