diff --git a/backend/app/data/sql_semantics.py b/backend/app/data/sql_semantics.py index bceb8f22..cc77c1d0 100644 --- a/backend/app/data/sql_semantics.py +++ b/backend/app/data/sql_semantics.py @@ -9161,8 +9161,168 @@ def output_names(sql: str, dialect: str | None = None, *, template: bool = False return [_output_name(i, read) for i in root.expressions] +# ========================================================================== +# 同一个形状（认证队列的「确定性重复」：条件的先后不算） +# ========================================================================== +# +# normalize_sql 的规范文本把字面量换成「?」，但条件照原文的先后：WHERE a = 1 AND b = 2 和 WHERE b = 2 AND a = 1 文本不同， +# 认证队列就拆成两组。这里在不认列的那一级规范形式（_light，和判等价同一套只保持语义的改写）上，把字面量抹成只看类型的 +# 占位，再把交换律成立的地方按抹掉之后的写法重排一遍：AND / OR 的各项、IN 列表、= / <> / + / * 的两边、比大小的两边（换边 +# 时大于小于跟着换）。规范形式里本来的排序看字面量的值（同一个形状换了参数，条件的先后可能就变了），所以要重排。重排时 +# 抹掉之后一样的几项按原值排：同一组里字面量的位置对得上。认不出的（解析不了、sqlglot 会读错的写法、规范化不了的结构） +# 为 None，调用方照旧按 normalize_sql 的文本分组：不同就是不同，只会分得细，不会把不同的并成一组。 + +#: 抹掉字面量之前给每个字面量记的编号（对回 normalize_sql 的字面量用） +_LIT_TAG = "sql_semantics_literal" +#: 对 normalize_sql 的先后时，数字字面量换成的标记：这个整数加编号（同 _NUMBER_SENTINEL，真实 SQL 里几乎不会写） +_LIT_NUMBER = 9_173_402_581_000 +_LIT_STRING = "__agl_l{}__" +_LIT_STRING_RE = re.compile(r"^'__agl_l(\d+)__'$") +#: 交换律成立、两边可以对调的运算 +_SYMMETRIC = ("EQ", "NEQ", "Add", "Mul", "NullSafeEQ") + + +@dataclass(frozen=True) +class SqlShape: + """sql_shape 的结果。key 一样就是同一个形状（只差字面量的值、条件的先后、交换律成立处两边的先后）；literals 是按规范 + 形式里的先后排的字面量（SQL 写法，字符串带引号），同一个形状的几条按位置对得上；positions[i] 是第 i 个在 + normalize_sql(sql).literals 里的位置（规范化时新算出来的字面量对不上，为 None）。""" + + key: str + literals: tuple[str, ...] + positions: tuple[int | None, ...] + + +def _normalized_order(root: Any, read: str | None) -> list[int] | None: + """normalize_sql 的字面量（按它重新生成的文本里的先后）各是哪一个字面量节点（_LIT_TAG 的编号）：把字面量换成带编号的 + 标记，同样地重新生成、切词，按先后读出编号。对不上（个数不同、有认不出的）为 None。""" + from sqlglot import exp + from sqlglot.optimizer.normalize_identifiers import normalize_identifiers + + marked = normalize_identifiers(root.copy(), dialect=read) + for lit in marked.find_all(exp.Literal): + tag = lit.meta.get(_LIT_TAG) + if tag is None: + return None + lit.set("this", _LIT_STRING.format(tag) if lit.is_string else str(_LIT_NUMBER + tag)) + text = marked.sql(dialect=read, comments=False) + out: list[int] = [] + for token in _tokens(text, read): + if not _is_literal(token): + continue + raw = text[token.start:token.end + 1] + found = _LIT_STRING_RE.match(raw) + if found: + out.append(int(found.group(1))) + continue + try: + out.append(int(raw) - _LIT_NUMBER) + except ValueError: + return None + return out + + +def _shape_print(node: Any, read: str | None) -> tuple[str, list[Any]]: + """规范形式的确定写法（同 _erased：字面量只写类型、不写值，按写出的先后收进列表），交换律成立的地方按抹掉之后的写法 + 重排（抹掉之后一样的几项按原值排）。""" + from sqlglot import exp + + def value_key(lits: list[Any]) -> tuple[str, ...]: + return tuple(lit.sql(dialect=read) for lit in lits) + + def ordered(parts: list[tuple[str, list[Any]]]) -> list[tuple[str, list[Any]]]: + return sorted(parts, key=lambda p: (p[0], value_key(p[1]))) + + if isinstance(node, exp.Paren): + return _shape_print(node.this, read) + if isinstance(node, exp.Literal): + return f"Literal({'s' if node.is_string else 'n'})", [node] + if isinstance(node, (exp.And, exp.Or)): + parts = ordered([_shape_print(p, read) for p in _flat(node, type(node))]) + return (f"{type(node).__name__}[{','.join(p[0] for p in parts)}]", + [lit for _, lits in parts for lit in lits]) + if isinstance(node, exp.Expr) and type(node).__name__ in _FLIP and not _quantified(node.this) \ + and not _quantified(node.expression): + left, right = _shape_print(node.this, read), _shape_print(node.expression, read) + name = type(node).__name__ + if (_fixed(node.this) and not _fixed(node.expression)) or ( + _fixed(node.this) == _fixed(node.expression) and (left[0], value_key(left[1])) > (right[0], value_key(right[1]))): + left, right, name = right, left, _FLIP[name] + return f"{name}({left[0]},{right[0]})", left[1] + right[1] + if isinstance(node, exp.Expr) and type(node).__name__ in _SYMMETRIC: + parts = ordered([_shape_print(node.this, read), _shape_print(node.expression, read)]) + return f"{type(node).__name__}({','.join(p[0] for p in parts)})", parts[0][1] + parts[1][1] + if isinstance(node, exp.In) and node.args.get("expressions"): + head = _shape_print(node.this, read) + parts = ordered([_shape_print(e, read) for e in node.expressions]) + rest = [f"{k}={_shape_print(v, read)[0]}" for k, v in sorted(node.args.items()) + if k not in ("this", "expressions") and v is not None and v is not False and v != []] + return (f"In({head[0]};[{','.join(p[0] for p in parts)}];{','.join(rest)})", + head[1] + [lit for _, lits in parts for lit in lits]) + if isinstance(node, exp.Expr): + pieces: list[str] = [] + lits: list[Any] = [] + for k, v in sorted(node.args.items()): + if v is None or v is False or v == []: + continue + text, found = _shape_print(v, read) + pieces.append(f"{k}={text}") + lits += found + mark = node.meta.get("sql_semantics_join") if isinstance(node, exp.Join) else None + if mark: + pieces.append(f"@join={mark}") + surface = node.meta.get(_SURFACE) + if surface: + pieces.append(f"@surface={surface}") + return f"{type(node).__name__}({','.join(pieces)})", lits + if isinstance(node, list): + parts = [_shape_print(v, read) for v in node] + return "[" + ",".join(p[0] for p in parts) + "]", [lit for _, found in parts for lit in found] + return repr(node), [] + + +def sql_shape(sql: str, dialect: str | None = None) -> SqlShape | None: + """一条查询的形状（见本节开头）：条件的先后、交换律成立处两边的先后、字面量的值都不算；别的不同一律算不同。认不出的为 + None（调用方按 normalize_sql 的文本分组）。从不抛异常。""" + from sqlglot import exp + + read = _read(dialect) + text = (_mysql_dashes(sql) if read == "mysql" and isinstance(sql, str) else sql or "").strip() + root, _ = _parse(text, read) + if root is None: + return None + try: + if _lost_in_parsing(text, read, root) or _respelled(text, read, root): + return None + for i, lit in enumerate(root.find_all(exp.Literal)): + lit.meta[_LIT_TAG] = i + order = _normalized_order(root, read) + if order is None or len(order) != len(normalize_sql(sql, dialect).literals): + return None + marked = _mark_surfaces(root, text, read) + if read == "sqlite": + _mark_comma_joins(root, text, read) + if not marked: + return None + try: + tree = _light(root, read, _Schema()).tree + except _Unsupported: + # 规范化不了的结构（递归 CTE、不确定的函数……）：只统一名字，条件的先后照原文 + tree = root.copy() + _normalize_names(tree, read, _Schema()) + printed, lits = _shape_print(tree, read) + except Exception: # noqa: BLE001 - 认形状本身的缺陷：当作认不出，调用方按文本分组 + logger.warning("认 SQL 的形状失败", exc_info=True) + return None + where = {tag: i for i, tag in enumerate(order)} + return SqlShape(hashlib.sha256(printed.encode()).hexdigest(), + tuple(lit.sql(dialect=read) for lit in lits), + tuple(where.get(lit.meta.get(_LIT_TAG)) if lit.meta.get(_LIT_TAG) is not None else None + for lit in lits)) + + __all__ = ["BINDING_LEVELS", "MAX_SQL_CHARS", "PLACEHOLDER", "Aggregate", "ColumnRef", "Computed", "FacetDiff", "Facets", - "JoinFacet", "LimitFacet", "NormalizedSql", "Part", "Projection", "SqlDiff", "TemplateBinding", + "JoinFacet", "LimitFacet", "NormalizedSql", "Part", "Projection", "SqlDiff", "SqlShape", "TemplateBinding", "TemplateMatch", "TemplateSlot", "TimePredicate", "bind_template", "catalog_labels", "diff_facets", "explain_sql_diff", "facets", "match_template", "normalize_sql", "output_names", "project_select", - "project_template", "render_bound"] + "project_template", "render_bound", "sql_shape"] diff --git a/backend/app/ledger/certify.py b/backend/app/ledger/certify.py index f4384c2b..3892a763 100644 --- a/backend/app/ledger/certify.py +++ b/backend/app/ledger/certify.py @@ -5,8 +5,11 @@ **认证队列**（GET /api/ledger/certify-queue，搭建者及以上）三种信号： - 确定性重复（shape）：问数据的探索运行里执行过、被报告引用（比对工件里标签记着的查询）、又没对上任何口径（none / related） - 的查询，按（数据源、normalize_sql 的规范文本）分组；出现在 ≥ REPEAT_RUNS 次运行、来自 ≥ REPEAT_CONVERSATIONS 段对话的进队。 - 跨运行变化的字面量位置是参数候选，不变的是口径常量。展示的是规范文本（字面量都换成了「?」，不带取值）； + 的查询，按（数据源、规范形）分组：sql_semantics.sql_shape 在判等价同一套规范形式上抹掉字面量、把 AND / OR 的各项、IN 列表、 + 交换律成立的比较两边重排——条件的先后不同、AND 两边换位的进同一组，条件真不同的不进；认不出规范形的按 normalize_sql 的 + 规范文本分组（只会分得细）。出现在 ≥ REPEAT_RUNS 次运行、来自 ≥ REPEAT_CONVERSATIONS 段对话的进队。跨运行变化的字面量 + 位置（按规范形里的先后对齐）是参数候选，不变的是口径常量。展示的是代表那条的规范文本（字面量都换成了「?」，不带取值）。 + 这一项的 id 按（数据源、规范形）算；以前按规范文本记的「不是口径」照样认（组里任何一条的规范文本对得上就算）； - 问题聚类（cluster）：app/chat/questions 全站唯一的聚类（只取问题原文、不含 history、排除系统派生的运行），≥ REPEAT_RUNS 个问题的类； - 被反复用的答案（reused）：导出过 ≥ 2 次（report.exported 审计，第 2 波）、做成过应用（蓝图的来源是这次运行）、在画布里 @@ -90,22 +93,38 @@ def item_id(kind: str, key: str) -> str: @dataclass(frozen=True) class Shape: - """一条查询的形状：数据源、规范文本（字面量换成「?」）、字面量（SQL 写法，字符串带引号）、规范化失败的原因。""" + """一条查询的形状：数据源、规范文本（字面量换成「?」）、字面量（SQL 写法，字符串带引号）、规范化失败的原因。 + + key 是分组用的规范形（sql_semantics.sql_shape，「shape:」开头；认不出时退回规范文本，「text:」开头）；canonical 是按 + 规范形里的先后排的字面量（同一组的几条按位置对得上，认哪些位置跨轮变了用它），positions[i] 是它在 literals 里的位置 + （规范化时新算出来的字面量为 None：没法写成模板参数）。""" source: str text: str literals: tuple[str, ...] problem: str | None = None + key: str = "" + canonical: tuple[str, ...] = () + positions: tuple[int | None, ...] = () + + def params(self, canonical_positions: Sequence[int]) -> list[int]: + """规范形里的位置 → literals 里的位置（template_of 要的）；对不上的不要。""" + return sorted(p for i in canonical_positions if i < len(self.positions) and (p := self.positions[i]) is not None) def shape_of(snap: Any, kinds: dict[str, str | None]) -> Shape | None: - from app.data.sql_semantics import normalize_sql + from app.data.sql_semantics import normalize_sql, sql_shape if not isinstance(snap, dict) or not isinstance(snap.get("sql"), str) or not snap["sql"].strip(): return None source = str(snap.get("source") or "") norm = normalize_sql(snap["sql"], kinds.get(source)) - return Shape(source, norm.text, norm.literals, norm.parse_error) + found = sql_shape(snap["sql"], kinds.get(source)) if not norm.parse_error else None + if found is None: + return Shape(source, norm.text, norm.literals, norm.parse_error, key=f"text:{norm.text}", + canonical=norm.literals, positions=tuple(range(len(norm.literals)))) + return Shape(source, norm.text, norm.literals, None, key=f"shape:{found.key}", canonical=found.literals, + positions=found.positions) def varying(literals: Sequence[Sequence[str]]) -> list[int]: @@ -174,11 +193,13 @@ def template_of(shape: Shape, params: Sequence[int], dialect: str | None, *, class _Member: run_id: str artifact: str + #: 按规范形里的先后排的字面量（Shape.canonical） literals: tuple[str, ...] conversation_id: str | None turn_id: str | None question: str related: list[dict[str, Any]] = field(default_factory=list) + shape: Shape | None = None def referenced_queries(content: dict[str, Any]) -> list[str]: @@ -297,31 +318,38 @@ async def queue(session: Any, *, principal: Any) -> dict[str, Any]: info = turns.get(run_id) or {} related = [{k: c.get(k) for k in ("caliber", "workflow_id", "version", "level")} for c in row.get("candidates") or [] if c.get("status") == "related"] - groups.setdefault((shape.source, shape.text), []).append(_Member( - run_id, query, shape.literals, info.get("conversation_id"), info.get("turn_id"), - str(info.get("question") or ""), related)) + groups.setdefault((shape.source, shape.key), []).append(_Member( + run_id, query, shape.canonical, info.get("conversation_id"), info.get("turn_id"), + str(info.get("question") or ""), related, shape)) return groups groups = await asyncio.to_thread(read) items: list[dict[str, Any]] = [] - for (source, text), members in groups.items(): + for (source, key), members in groups.items(): once = list({m.run_id: m for m in members}.values()) conversations = {m.conversation_id or f"run:{m.run_id}" for m in once} if len(once) < REPEAT_RUNS or len(conversations) < REPEAT_CONVERSATIONS: continue - iid = item_id("shape", f"{source}\n{text}") - if iid in dismissed: + iid = item_id("shape", f"{source}\n{key}") + # 以前按规范文本记的「不是口径」：组里任何一条的规范文本对得上就照样不进队 + legacy = {item_id("shape", f"{source}\n{m.shape.text}") for m in once if m.shape is not None} + if iid in dismissed or legacy & dismissed: continue - params = varying([m.literals for m in once]) - related = list({r["caliber"]: r for m in once for r in m.related if r.get("caliber")}.values()) rep = once[0] + shape = rep.shape + if shape is None: + continue + varied = varying([m.literals for m in once]) + params = [(i, p) for i in varied if i < len(shape.positions) and (p := shape.positions[i]) is not None] + related = list({r["caliber"]: r for m in once for r in m.related if r.get("caliber")}.values()) items.append({ - "id": iid, "kind": "shape", "source": source, "shape": text, "runs": len(once), + "id": iid, "kind": "shape", "source": source, "shape": shape.text, "runs": len(once), "conversations": len(conversations), "question": rep.question, "turns": [_turn_view(m) for m in once[:TURNS_LIMIT]], - "params": [{"index": i, "distinct": len({m.literals[i] for m in once if len(m.literals) > i})} - for i in params], - "constants": len(rep.literals) - len(params), "related": related, + # index 是代表那条规范文本里的第几个「?」；distinct 按规范形对齐了数 + "params": [{"index": p, "distinct": len({m.literals[i] for m in once if len(m.literals) > i})} + for i, p in sorted(params, key=lambda ip: ip[1])], + "constants": len(shape.literals) - len(params), "related": related, "representative": {"run_id": rep.run_id, "query": rep.artifact}, "peers": [m.run_id for m in once[1:]], "frozen_into": _frozen_of([m.run_id for m in once], frozen), @@ -447,7 +475,7 @@ async def freeze(session: Any, run_id: str, *, peers: Sequence[str] = (), name: raise CertifyRefused(409, NO_QUERIES) kinds = {n: k for n, k in (await session.execute(select(DataSource.name, DataSource.kind))).all()} - # 同组的别的运行：它们封存的查询里同形状的那一条的字面量 + # 同组的别的运行：它们封存的查询里同形状（规范形相同，见 Shape.key）的那一条的字面量，按规范形里的先后 peer_literals: dict[tuple[str, str], list[tuple[str, ...]]] = {} peer_sources: list[dict[str, Any]] = [] peer_runs = [r for r in [await session.get(Run, p) for p in dict.fromkeys(peers) if p != run_id] if r is not None] @@ -460,7 +488,7 @@ async def freeze(session: Any, run_id: str, *, peers: Sequence[str] = (), name: except ValueError: continue if shape is not None and not shape.problem: - peer_literals.setdefault((shape.source, shape.text), []).append(shape.literals) + peer_literals.setdefault((shape.source, shape.key), []).append(shape.canonical) sealed = await _sealed(run_id) shapes: list[Shape] = [] snaps: list[dict[str, Any]] = [] @@ -485,7 +513,7 @@ async def freeze(session: Any, run_id: str, *, peers: Sequence[str] = (), name: metrics: list[dict[str, Any]] = [] multi = False for i, (shape, snap) in enumerate(zip(shapes, snaps)): - params = varying([shape.literals, *peer_literals.get((shape.source, shape.text), [])]) + params = shape.params(varying([shape.canonical, *peer_literals.get((shape.source, shape.key), [])])) template, new_fields = template_of(shape, params, kinds.get(shape.source), start=len(fields) + 1) fields += new_fields nodes.append({"id": nid(i), "type": "tool", "position": {"x": 0, "y": 0}, diff --git a/backend/tests/test_ledger_certify.py b/backend/tests/test_ledger_certify.py index 4444064c..1431921c 100644 --- a/backend/tests/test_ledger_certify.py +++ b/backend/tests/test_ledger_certify.py @@ -270,3 +270,80 @@ async def test_freezing_needs_a_matched_successful_run(client, monkeypatch): row = await S.explore(monkeypatch, S.ELSEWHERE, text="一共几家店，没有引用。") res = await client.post("/api/copilot/from-run", json={"run_id": row.id, "freeze": True}) assert res.status_code == 409 and res.json()["detail"] == certify.NO_QUERIES + + +# -------------------------------------------------------------------------- +# 确定性重复按规范形分组：条件的先后不算，条件真不同的不算一组 +# -------------------------------------------------------------------------- + +#: 同一个形状的三种写法：条件换序、AND 两边换位、比较两边换位 +REORDERED = [ + "SELECT SUM(amount) AS gmv FROM orders WHERE status = 'paid' AND paid_at >= '{s}' AND paid_at < '{e}'", + "SELECT SUM(amount) AS gmv FROM orders WHERE paid_at < '{e}' AND paid_at >= '{s}' AND status = 'paid'", + "SELECT SUM(amount) AS gmv FROM orders WHERE '{s}' <= paid_at AND ('paid' = status AND paid_at < '{e}')", +] + + +def test_conditions_in_another_order_are_the_same_shape_and_their_literals_line_up(): + from app.data.sql_semantics import sql_shape + + shapes = [sql_shape(sql.format(s=s, e=e), "sqlite") for sql, (s, e) in zip(REORDERED, WEEKS)] + assert all(sh is not None for sh in shapes) + assert len({sh.key for sh in shapes}) == 1 + # 按规范形里的先后对齐：同一个位置是同一个条件里的那个字面量——变了的是两个日期，'paid' 在每一条里都落在同一个位置 + assert len(certify.varying([sh.literals for sh in shapes])) == 2 + assert len({sh.literals.index("'paid'") for sh in shapes}) == 1 + starts = {sh.literals.index(f"'{s}'") for sh, (s, _) in zip(shapes, WEEKS)} + assert len(starts) == 1, "起始日期在每一条里都落在同一个位置" + + +@pytest.mark.parametrize("other", [ + "SELECT SUM(amount) AS gmv FROM orders WHERE status = 'paid' AND created_at >= '2026-09-01' AND paid_at < '2026-09-08'", + "SELECT SUM(amount) AS gmv FROM orders WHERE status = 'paid' OR paid_at >= '2026-09-01' AND paid_at < '2026-09-08'", + "SELECT SUM(amount) AS gmv FROM orders WHERE status = 'paid' AND paid_at >= '2026-09-01'", + "SELECT SUM(amount) AS gmv FROM orders WHERE status = 'paid' AND paid_at > '2026-09-01' AND paid_at < '2026-09-08'", + "SELECT SUM(amount) AS gmv FROM orders WHERE status = 1 AND paid_at >= '2026-09-01' AND paid_at < '2026-09-08'", + "SELECT SUM(amount) AS gmv FROM orders WHERE status IN ('paid', 'open') AND paid_at >= '2026-09-01' " + "AND paid_at < '2026-09-08'", + "SELECT COUNT(*) AS gmv FROM orders WHERE status = 'paid' AND paid_at >= '2026-09-01' AND paid_at < '2026-09-08'", +], ids=["other-column", "or", "missing-condition", "other-operator", "number-not-string", "in-list", "other-aggregate"]) +def test_really_different_conditions_are_different_shapes(other): + from app.data.sql_semantics import sql_shape + + base = sql_shape(REORDERED[0].format(s="2026-09-01", e="2026-09-08"), "sqlite") + found = sql_shape(other, "sqlite") + assert base is not None and found is not None and found.key != base.key + + +async def test_queue_groups_reordered_conditions_and_freezing_aligns_their_parameters(client, monkeypatch): + a, b = await conversation(), await conversation() + runs = [await ask(monkeypatch, conv, sql.format(s=s, e=e)) + for conv, sql, (s, e) in zip((a, a, b), REORDERED, WEEKS)] + # 条件真不同的（多筛了一个地区）不进这一组 + other = await ask(monkeypatch, b, "SELECT SUM(amount) AS gmv FROM orders WHERE status = 'paid' AND region = 'east' " + "AND paid_at >= '2026-09-01' AND paid_at < '2026-09-08'") + [item] = shape_items((await client.get("/api/ledger/certify-queue")).json()) + assert item["runs"] == 3 and {t["run_id"] for t in item["turns"]} == {r.id for r in runs} + assert other.id not in {t["run_id"] for t in item["turns"]} + assert len(item["params"]) == 2 and all(p["distinct"] == 3 for p in item["params"]) and item["constants"] == 1 + + rep = item["representative"]["run_id"] + res = await client.post("/api/copilot/from-run", json={"run_id": rep, "freeze": True, "peers": item["peers"]}) + assert res.status_code == 200, res.text + made = res.json() + assert made["params"] == ["p1", "p2"], "条件换了序的几轮照样认得出两个日期是参数" + async with SessionLocal() as session: + wf = await session.get(Workflow, made["workflow_id"]) + sql = {n["id"]: n for n in wf.graph["nodes"]}["fetch"]["data"]["config"]["args"]["sql"] + assert "'paid'" in sql and "'{{ input.p1 }}'" in sql and "'{{ input.p2 }}'" in sql + + +async def test_a_not_a_caliber_recorded_by_the_old_text_id_still_counts(client, monkeypatch): + """以前「不是口径」按规范文本算的 id 记：分组改按规范形之后，组里任何一条的规范文本对得上就照样不进队。""" + await three_asks(monkeypatch) + [item] = shape_items((await client.get("/api/ledger/certify-queue")).json()) + old = certify.item_id("shape", f"{S.CURRENT['source']}\n{item['shape']}") + assert old != item["id"] + res = await client.post("/api/ledger/certify-queue/dismiss", json={"item_id": old, "kind": "shape"}) + assert res.status_code == 200 + assert shape_items((await client.get("/api/ledger/certify-queue")).json()) == []