diff --git a/backend/app/data/sql_semantics.py b/backend/app/data/sql_semantics.py index cc77c1d0..ecf808e2 100644 --- a/backend/app/data/sql_semantics.py +++ b/backend/app/data/sql_semantics.py @@ -14,6 +14,10 @@ A / B,只差一个方面、又对得上数据目录里的事实(时间归属 (规范形式一字不差)才返回空列表。 - facets:把一条查询拆成表、连接、WHERE 合取式、时间条件、分组、聚合、算出来的列、HAVING、取哪几行(给调用方取 结构用;CTE、子查询按位置编号,见 _Statement)。facets 不受下面等价判定的改动影响。 +- bind_template / match_template:口径模板的参数绑定与等价证明(口径账本,见那一节)。 +- project_select / project_template:按引用的列投影——去掉最外层选择列表里没被引用、去掉也不改变被引用那几列取值的 + 选择项,认不准的一律不投影(见那一节)。投影本身不判等价,投影后的 SQL 照样交给上面的判等价。 +- sql_shape:同一个形状(认证队列的确定性重复):条件的先后、交换律成立处两边的先后、字面量的值都不算(见那一节)。 **等价怎么判。** 过去几轮按方面归纳任意 SQL,每轮修好一批又冒出一批(还回退过),收不拢;现在先判等价、再说差在哪。 两边各自规范化,规范化的每一步都只做保持语义的变换,同一级的两个规范形式一字不差(比语法树逐个节点的类型和参数, @@ -9245,8 +9249,9 @@ def _shape_print(node: Any, read: str | None) -> tuple[str, list[Any]]: 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]))): + later = (left[0], value_key(left[1])) > (right[0], value_key(right[1])) + fixed_left, fixed_right = _fixed(node.this), _fixed(node.expression) + if (fixed_left and not fixed_right) or (fixed_left == fixed_right and later): 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: diff --git a/backend/app/engine/lineage.py b/backend/app/engine/lineage.py index 21302fd5..647a4e69 100644 --- a/backend/app/engine/lineage.py +++ b/backend/app/engine/lineage.py @@ -97,8 +97,9 @@ def metric_copy(path: str) -> tuple[str, str, int] | None: if not (isinstance(tree, ast.Attribute) and tree.attr == "value" and isinstance(tree.value, ast.Subscript)): return None index, metrics = tree.value.slice, tree.value.value - if not (isinstance(metrics, ast.Attribute) and metrics.attr == "metrics" and isinstance(metrics.value, ast.Attribute) - and isinstance(metrics.value.value, ast.Name) and metrics.value.value.id in ("vars", "nodes")): + if not (isinstance(metrics, ast.Attribute) and metrics.attr == "metrics" + and isinstance(metrics.value, ast.Attribute) and isinstance(metrics.value.value, ast.Name) + and metrics.value.value.id in ("vars", "nodes")): return None if not (isinstance(index, ast.Constant) and type(index.value) is int and index.value >= 0): return None diff --git a/backend/app/engine/nodes/metrics.py b/backend/app/engine/nodes/metrics.py index f19799ea..740d545b 100644 --- a/backend/app/engine/nodes/metrics.py +++ b/backend/app/engine/nodes/metrics.py @@ -656,8 +656,10 @@ def _input_of(path: str, value: Any, state: GraphState, ctx: NodeContext, # 沙箱算出来的数喂给口径卡:记下它自认是「取数」还是「计算」,治理规则按它判 entry["role"] = str(producer.config.get("evidence_role") or "compute") entry["status"] = "ok" if value is not None else "missing" - if producer is not None and str(producer.type) == "metrics" and (up := _upstream(path, tctx or template_context(state))): - entry["upstream"] = up + if producer is not None and str(producer.type) == "metrics": + # 转抄另一张口径卡的指标:跨卡往下追(engine/lineage.metric_copy) + if (up := _upstream(path, tctx or template_context(state))) is not None: + entry["upstream"] = up if producer is not None and str(producer.type) == "agent" and field: cited = _agent_field(state, producer.id, root if head else "", field) if cited is not None: diff --git a/backend/app/ledger/certify.py b/backend/app/ledger/certify.py index 049c9ba7..3f1d454c 100644 --- a/backend/app/ledger/certify.py +++ b/backend/app/ledger/certify.py @@ -111,7 +111,8 @@ class Shape: 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) + found = (self.positions[i] for i in canonical_positions if i < len(self.positions)) + return sorted(p for p in found if p is not None) def shape_of(snap: Any, kinds: dict[str, str | None]) -> Shape | None: @@ -258,7 +259,8 @@ async def _dismissed_members(session: Any) -> list[set[str]]: def _cluster_dismissed(cluster: Any, dismissed: set[str], members: list[set[str]]) -> bool: """这一类点过「不是口径」没有:类的 id 记过;以前按种子问题原文算的 id 记过;或者过半的成员是某一次记下的。""" - if f"cluster:{cluster.id}" in dismissed or item_id("cluster", " ".join(cluster.members[0].text.split())) in dismissed: + legacy = item_id("cluster", " ".join(cluster.members[0].text.split())) + if f"cluster:{cluster.id}" in dismissed or legacy in dismissed: return True keys = set(cluster.keys) return any(len(keys & seen) * 2 > len(keys) for seen in members) diff --git a/backend/app/ledger/labels.py b/backend/app/ledger/labels.py index 5ded531b..14be0c3b 100644 --- a/backend/app/ledger/labels.py +++ b/backend/app/ledger/labels.py @@ -431,5 +431,5 @@ def related_view(item: Mapping[str, Any], *, hide: Callable[[Any], bool], shown: __all__ = ["CERTIFIED_NOTE", "HOWS", "KINDS", "LEVEL_TEXT", "POST_SEAL_NOTE", "RETRO_NOTE", "binding_view", "caliber_key", - "card_expressions", "card_reads", "cell_label", "diff_view", "label_doc", "metric_label", "referenced_columns", - "related_view", "view_label", "whole_of"] + "card_expressions", "card_reads", "cell_label", "diff_view", "label_doc", "metric_label", + "referenced_columns", "related_view", "view_label", "whole_of"] diff --git a/backend/app/ledger/match.py b/backend/app/ledger/match.py index 4b77fc05..2a128b40 100644 --- a/backend/app/ledger/match.py +++ b/backend/app/ledger/match.py @@ -513,13 +513,10 @@ def compare_queries(queries: list[Query], calibers: list[Caliber], *, deadline: return out -def compute(sealed: Any, calibers: list[Caliber], kinds: dict[str, str | None], *, timeout: float) -> dict[str, Any]: - """一次运行的比对工件内容(见 ledger/store 的模块说明)。在线程里调:读快照、比 SQL、标数都是同步的活。""" - from app.api.evidence import _choose, _reports +def _card_loader(sealed: Any) -> Any: + """按工件读封存范围内的指标集(不在封存台账里的、读不出的为 None)。""" from app.core.artifact_store import load - deadline = time.monotonic() + timeout - queries = read_queries(sealed, kinds) sealed_cards = {artifact for _, artifact in sealed.ledger} def load_card(artifact: str) -> dict[str, Any] | None: @@ -531,6 +528,25 @@ def compute(sealed: Any, calibers: list[Caliber], kinds: dict[str, str | None], return None return card if isinstance(card, dict) else None + return load_card + + +def referenced_of(sealed: Any) -> dict[str, set[str]]: + """这次运行封存的报告引用了哪条查询的哪几列(labels.referenced_columns,只读哈希核对一致的报告):compare_queries 的 + referenced。评测逐题比 SQL 时传它,就和账本一样按引用的列投影。在线程里调(读盘)。""" + from app.api.evidence import _reports + + docs = [r.doc for r in _reports(sealed) if r.doc is not None and r.hash_ok is True] + return ledger_labels.referenced_columns(docs, _card_loader(sealed)) + + +def compute(sealed: Any, calibers: list[Caliber], kinds: dict[str, str | None], *, timeout: float) -> dict[str, Any]: + """一次运行的比对工件内容(见 ledger/store 的模块说明)。在线程里调:读快照、比 SQL、标数都是同步的活。""" + from app.api.evidence import _choose, _reports + + deadline = time.monotonic() + timeout + queries = read_queries(sealed, kinds) + load_card = _card_loader(sealed) found = _reports(sealed) readable = [report for report in found if report.doc is not None and report.hash_ok is True] if not sealed.trusted: @@ -743,7 +759,7 @@ async def shutdown() -> None: await asyncio.gather(*tasks, return_exceptions=True) -__all__ = ["BACKFILL_DAYS", "CONCURRENCY", "EXPLORE_ORIGINS", "EXTRA_COLUMNS", "MAX_PROJECTIONS", "TIMEOUT_S", "Caliber", - "Fetch", "Query", "backfill", "caliber_of", "compare_queries", "composed_nodes", "compute", "drain", - "ineligible", "load_calibers", "match_run", "pending", "read_queries", "schedule", "shutdown", - "start_backfill", "wanted"] +__all__ = ["BACKFILL_DAYS", "CONCURRENCY", "EXPLORE_ORIGINS", "EXTRA_COLUMNS", "MAX_PROJECTIONS", "TIMEOUT_S", + "Caliber", "Fetch", "Query", "backfill", "caliber_of", "compare_queries", "composed_nodes", "compute", + "drain", "ineligible", "load_calibers", "match_run", "pending", "read_queries", "referenced_of", + "schedule", "shutdown", "start_backfill", "wanted"] diff --git a/backend/tests/test_ledger_projection.py b/backend/tests/test_ledger_projection.py index 4a5cb25c..869dff52 100644 --- a/backend/tests/test_ledger_projection.py +++ b/backend/tests/test_ledger_projection.py @@ -193,6 +193,10 @@ async def test_last_weeks_sales_with_extra_columns_is_whole_equivalent(engine_up content = (await store.latest(row.id)).content [q] = content["queries"] assert q["status"] == "proven" + # 评测逐题比 SQL 时拿同一份「引用了哪几列」(referenced_of),和账本一样投影 + from app.api.evidence import _sealed + + assert ledger_match.referenced_of(await _sealed(row.id)) == {q["artifact"]: {"销售额"}} [cand] = q["candidates"] assert cand["projection"]["dropped"] == 3 and cand["columns"] == [{"index": 0, "ours": "销售额", "caliber": "gmv"}] [labels] = [r["labels"] for r in content["reports"].values()]