diff --git a/backend/app/bundle/build.py b/backend/app/bundle/build.py index f742a610..61219418 100644 --- a/backend/app/bundle/build.py +++ b/backend/app/bundle/build.py @@ -498,9 +498,10 @@ def _snapshot_hidden(content: dict[str, Any], source_masks: Mapping[str, Any]) - # -------------------------------------------------------------------------- #: 数据源查询工具（tools/datasource.py）落的查询快照里会有的键。有别的键的（比如核查 agent 在查询时就把遮罩列换成 -#: 「已遮罩」、另记下了遮了哪些列）不知道它记着什么，照现有规则只给 id +#: 「已遮罩」、另记下了遮了哪些列）不知道它记着什么，照现有规则只给 id。no_data 是查询区间里这张表没有数据的那几列 +#: （data/coverage.py）：只有日期和表名、列名，遮罩列上的不探查也不记 _QUERY_KEYS = frozenset({"columns", "rows", "row_count", "truncated", "elapsed_ms", "sql", "column_types", "source", - "checks", "schema_artifact", "mask_columns", "data_version"}) + "checks", "schema_artifact", "mask_columns", "data_version", "no_data"}) #: 导出时只知道数据源设的遮罩、不知道是哪种库：SQL 按每种支持的库的读法各解析一遍（None 是 sqlglot 的通用读法）， #: 至少一种读得通，读得通的每一种都要过关 _READINGS = (None, "sqlite", "mysql", "postgres", "oracle") diff --git a/backend/app/data/coverage.py b/backend/app/data/coverage.py new file mode 100644 index 00000000..eeb147b6 --- /dev/null +++ b/backend/app/data/coverage.py @@ -0,0 +1,785 @@ +"""查询区间里这张表有没有数据：运行时现查（查询结果整列是 0、助手搭图自查时用）。 + +**为什么要有它。** 部署上撞过：一份周报按 `event_date >= DATE_SUB(CURRENT_DATE, INTERVAL (WEEKDAY(CURRENT_DATE)+7) DAY) +AND event_date < …` 取上周、上上周的到访人数，那张表却只存最新一天（每天整表重灌），6 列全是 0，口径卡算出一排 0 和 0%， +周报照写「上周到访人数为 0 人」「周环比 0.00%」。这些 0 不是「没有到访」，是这段时间表里根本没有数据。剖析记过日期列的 +最小、最大值，但对滚动表不可靠（剖析那天表里是前一天的数据，运行时已经换成当天），所以要运行时现查。 + +**做什么。** 两个入口共用一套： + +- result_gaps：一次查询的结果里有「整列都是 0 或空」的数值列时（结果一行都没有的不归这里管），对每个这样的列，把 SQL + 收窄到算出这一列的那一项（去掉别的选择项、分组、HAVING、排序、取行数，同双盲复算差异说明的「收窄到这一格」，这里只做 + 分析、不执行），取出它用到的表和时间谓词（列、比较符、界）； +- sql_gaps：助手搭图自查时，工具节点里写死的 SQL 还没跑、没有结果：对 SQL 里所有带时间谓词的表都查。 + +界是字面量的直接解析；按运行当天现算的（不含列引用的表达式：CURRENT_DATE、DATE_SUB(…)、date('now', …)……）用同一个数据源 +跑一条 `SELECT 界1, 界2 …` 在库里算出来（Oracle 加 FROM DUAL）。这一列在这张表里的实际范围现查 `SELECT MIN(列), MAX(列) +FROM 表`（每条探查 PROBE_TIMEOUT_S 秒）；超时或失败退回剖析、取值观察记下的范围（facts，照实标「按 X 日的剖析」），都没有 +就不判。日期按常见写法归一（DATE / DATETIME、'YYYY-MM-DD'、'YYYYMMDD'、8 位整数），认不出就不判；列是文本、界也是文本时 +两边写法要一样（'20261012' 对 '2026-10-12' 按字符串比，结果和按日期比对不上），不一样的不判。区间和实际范围完全不重叠才算 +「没有数据」；一列用到几段区间（本周减上周）时每一段都不重叠才算。 + +**和 sqlcheck 一样宁可漏报。** 所有探查都走数据源查询同一道只读守卫（data/guard.check + is_write），遮罩列上的不探查、不报 +值；总耗时有上限（BUDGET_S）；解析不了、认不出、探查失败一律当没问题，出错不影响查询本身（同 tools/datasource._sql_checks）。 +""" +from __future__ import annotations + +import asyncio +import logging +import re +import time +from collections.abc import Awaitable, Callable, Iterable, Mapping +from dataclasses import dataclass, field +from datetime import date, datetime, timedelta +from typing import Any + +from app.data.names import name_key + +logger = logging.getLogger(__name__) + +#: 一次检查（一份查询结果、一条自查的 SQL）所有探查加起来最多多少秒 +BUDGET_S = 20.0 +#: 每条探查查询的时限 +PROBE_TIMEOUT_S = 10.0 +#: 一次最多探查几张表的范围、几条按运行当天现算的界：再多就是 SQL 太复杂，不为它把库查一遍 +MAX_RANGES = 8 +MAX_BOUNDS = 24 +#: 太长的 SQL 不分析（同 sqlcheck.MAX_SQL_CHARS） +MAX_SQL_CHARS = 20_000 +#: 一条 no_data 里最多列几个结果列（计数照实） +MAX_COLUMNS = 12 + +#: 给人和模型看的说法 +NO_DATA_MESSAGE = "{columns}查的是 {window}，但 {table} 的 {column} {data}：这些 0 不代表这段时间真的是 0，是这段时间表里没有数据{basis}" +NO_DATA_SQL_MESSAGE = "查的区间是 {window}，但 {table} 的 {column} {data}{basis}" +BASIS_PROFILE = "（现查表的范围没有成功，按 {day} 的剖析）" +BASIS_OBSERVED = "（现查表的范围没有成功，按 {day} 的取值观察）" +#: 写进口径卡、写作目录、双盲复算原因的那半句 +NO_DATA_SHORT = "查询区间里这张表没有数据" + +#: (表达式的类型, 方言) → 按运行当天现算的界：不认列、不认子查询，函数、字面量、INTERVAL 都可以 +_COMPARE = {"GT": ">", "GTE": ">=", "LT": "<", "LTE": "<=", "EQ": "="} +_FLIP = {">": "<", ">=": "<=", "<": ">", "<=": ">=", "=": "="} +_ISO = re.compile(r"^(\d{4})-(\d{1,2})-(\d{1,2})(?:[ T](\d{1,2}):(\d{2})(?::(\d{2})(?:\.\d+)?)?)?" + r"(?:Z|[+-]\d{2}:?\d{2})?$") +_COMPACT = re.compile(r"^(\d{4})(\d{2})(\d{2})$") + + +# ========================================================================== +# 日期的归一 +# ========================================================================== + + +@dataclass(frozen=True) +class _Day: + """归一后的一个日期：day 是哪一天，timed 是带不带一天里的时刻（00:00:00 不算），shape 是写法（iso / compact / + number / date：库交回的是日期类型），kind 是库交回的类型（date / datetime / text / number，字面量按它的写法）。""" + + day: date + timed: bool + shape: str + kind: str + + +def _as_day(value: Any, kind: str | None = None) -> _Day | None: + """一个值归一成日期；认不出返回 None。kind 是库交回的列类型（run_query 的 column_types），字面量不给。""" + if isinstance(value, bool) or value is None: + return None + if isinstance(value, datetime): + return _Day(value.date(), bool(value.time().replace(microsecond=0) != datetime.min.time()), "date", "datetime") + if isinstance(value, date): + return _Day(value, False, "date", "date") + if isinstance(value, (int, float)): + if isinstance(value, float) and not value.is_integer(): + return None + found = _COMPACT.match(str(int(value))) + day = _ymd(found.group(1), found.group(2), found.group(3)) if found else None + return _Day(day, False, "number", kind or "number") if day else None + if not isinstance(value, str): + return None + text = value.strip() + if found := _COMPACT.match(text): + day = _ymd(found.group(1), found.group(2), found.group(3)) + return _Day(day, False, "compact", kind or "text") if day else None + if found := _ISO.match(text): + day = _ymd(found.group(1), found.group(2), found.group(3)) + if day is None: + return None + timed = any(int(found.group(i) or 0) for i in (4, 5, 6)) + shape = "date" if kind in ("date", "datetime") else "iso" + return _Day(day, timed, shape, kind or "text") + return None + + +def _ymd(y: str, m: str, d: str) -> date | None: + try: + return date(int(y), int(m), int(d)) + except ValueError: + return None + + +def _comparable(column: _Day, bound: _Day, *, bare: bool, read: str | None) -> bool: + """库里拿这一列和这个界比，是不是按日期在比（我们的判断才对得上）： + + - 列外面套着函数（DATE(x)、STR_TO_DATE(x, …)）：比的是函数的结果，按日期算； + - 列本身是日期类型：库把文本的界转成日期再比； + - 界是库算出来的日期类型（DATE_SUB(CURRENT_DATE, …)）、列是文本：MySQL 把文本转成日期再比，别的库不一定（SQLite 的 + date('now') 交回的是文本，照下一条比写法），只认 MySQL； + - 两边都是文本、都是数：写法得一样（'20261012' 对 '2026-10-12' 按字符串比，和按日期比对不上）。""" + if not bare or column.kind in ("date", "datetime"): + return True + if bound.kind in ("date", "datetime"): + return read == "mysql" and column.kind == "text" + if column.kind == "text" and bound.kind == "text": + return column.shape == bound.shape + return column.kind == bound.kind == "number" and column.shape == bound.shape == "number" + + +# ========================================================================== +# 从 SQL 里认时间谓词 +# ========================================================================== + + +@dataclass +class _Filter: + """一段 SELECT 里对一张实表的一列的时间条件：table / column 是 sqlglot 的节点（探查时照它写 SQL），conds 是 + [(比较符, 界的节点, 列是不是光着的)]。key 按内容认（表、列、各个条件的写法）：几列各自收窄出来的同一段条件是同一个， + 判一次。""" + + key: tuple[Any, ...] + table: Any + table_name: str + column: Any + column_name: str + conds: list[tuple[str, Any, bool]] = field(default_factory=list) + + +def _read(kind: str | None) -> str | None: + from app.data.sqlcheck import dialect_of + + return dialect_of(kind) + + +def _parse(sql: str, read: str | None) -> Any: + import sqlglot + from sqlglot import exp + + if not isinstance(sql, str) or not sql.strip() or len(sql) > MAX_SQL_CHARS or "{{" in sql: + return None + try: + statements = [s for s in sqlglot.parse(sql, read=read) if s is not None] + except Exception: # noqa: BLE001 - 解析不了就不判 + return None + if len(statements) != 1 or not isinstance(statements[0], exp.Query): + return None + return statements[0] + + +def _output_name(item: Any, read: str | None) -> str: + from sqlglot import exp + + if isinstance(item, exp.Alias): + return item.alias + if isinstance(item, exp.Column): + return item.name + return item.sql(dialect=read) + + +def _same_name(a: str, b: str) -> bool: + return "".join(a.split()).lower() == "".join(b.split()).lower() + + +def narrowed(tree: Any, column: str, read: str | None) -> Any: + """收窄到算出 column 这一列的那一项：最外层 SELECT 只留它，分组、HAVING、排序、取行数去掉（只拿来认时间谓词，不执行， + 不用像双盲复算那样按这一行的分组值补等值条件）。认不准的（不是一段 SELECT、SELECT *、对不上或对上好几项）交回整条。""" + from sqlglot import exp + + if not isinstance(tree, exp.Select): + return tree + items = list(tree.expressions) + if any(isinstance(i, exp.Star) or (isinstance(i, exp.Column) and isinstance(i.this, exp.Star)) for i in items): + return tree + hits = [i for i in items if _same_name(_output_name(i, read), column)] + if len(hits) != 1: + return tree + out = tree.copy() + out.set("expressions", [hits[0].copy()]) + for arg in ("group", "having", "order", "limit", "offset", "qualify"): + out.set(arg, None) + return out + + +def _columns_in(node: Any) -> list[Any]: + from sqlglot import exp + + return [c for c in node.find_all(exp.Column) if not isinstance(c.this, exp.Star)] + + +def _plain_bound(node: Any) -> bool: + """界里没有列、子查询、聚合、窗口：字面量，或者按运行当天现算的表达式。""" + from sqlglot import exp + + return not _columns_in(node) and not any( + isinstance(n, (exp.Select, exp.Subquery, exp.AggFunc, exp.Window, exp.Placeholder, exp.Parameter)) + for n in node.walk()) + + +def _timeish(subject: Any, column: Any, bounds: list[Any]) -> bool: + """像是时间条件：列名像日期、时间，或者列套着日期函数、界里有日期函数、像日期的常量、按运行当天现算的写法。""" + from app.data import sql_semantics as ss + + name = str(column.name or "") + return bool(ss._temporal_name(name) or ss._timeish_name(name) or ss._datey(subject) + or any(ss._datey(b) or ss._date_like_literal(b) or ss._now_based(b) for b in bounds)) + + +def _condition(cond: Any) -> tuple[Any, Any, list[tuple[str, Any]], bool] | None: + """一个条件是不是「一列 比 界」：(列, 列那一侧的表达式, [(比较符, 界)], 列是不是光着的)。BETWEEN、IN（常量列表） + 拆成几项；NOT、子查询、两边都有列的不算。""" + from sqlglot import exp + + from app.data.sqlcheck import _bare_column, _unparen + + n = _unparen(cond) + if n.args.get("negate"): + return None + if isinstance(n, exp.Between): + subject, pairs = n.this, [(">=", n.args.get("low")), ("<=", n.args.get("high"))] + elif isinstance(n, exp.In) and n.expressions and not n.args.get("query"): + subject, pairs = n.this, [("in", e) for e in n.expressions] + elif type(n).__name__ in _COMPARE and isinstance(n, exp.Binary): + op = _COMPARE[type(n).__name__] + left, right = n.this, n.expression + if _columns_in(left) and not _columns_in(right): + subject, pairs = left, [(op, right)] + elif _columns_in(right) and not _columns_in(left): + subject, pairs = right, [(_FLIP[op], left)] + else: + return None + else: + return None + if subject is None or any(b is None or not _plain_bound(b) for _, b in pairs): + return None + cols = _columns_in(subject) + if len(cols) != 1 or any(isinstance(x, (exp.AggFunc, exp.Select, exp.Window)) for x in subject.walk()): + return None + return cols[0], subject, pairs, _bare_column(subject) is not None + + +def _real_tables(select: Any, ctes: set[str]) -> dict[str, Any]: + """这一段 FROM / JOIN 里的实表：{别名或表名的 name_key: 表节点}。CTE、子查询不算。""" + from sqlglot import exp + + sources = [select.args.get("from_") or select.args.get("from")] + list(select.args.get("joins") or []) + out: dict[str, Any] = {} + for source in sources: + table = getattr(source, "this", None) + if isinstance(table, exp.Table) and table.name and name_key(table.name) not in ctes: + out[name_key(table.alias_or_name)] = table + return out + + +def _schema_columns(schema: Mapping[str, Any] | None, table: Any) -> set[str] | None: + """表结构里这张表的列（name_key）；表结构里没有这张表为 None。""" + tables = (schema or {}).get("tables") if isinstance(schema, Mapping) else None + if not isinstance(tables, Mapping): + return None + for name, meta in tables.items(): + if name_key(str(name).rsplit(".", 1)[-1]) == name_key(table.name) and isinstance(meta, Mapping): + return {name_key(str(c.get("name"))) for c in meta.get("columns") or [] if isinstance(c, Mapping)} + return None + + +def filters(tree: Any, schema: Mapping[str, Any] | None = None) -> list[_Filter]: + """语法树里各段 SELECT 的 WHERE 中，对实表的一列的时间条件，按（段、表、列）归在一起。""" + from sqlglot import exp + + from app.data.sqlcheck import _conjuncts + + ctes = {name_key(c.alias_or_name) for c in tree.find_all(exp.CTE) if c.alias_or_name} + found: dict[tuple[int, str, str], _Filter] = {} + for index, select in enumerate(tree.find_all(exp.Select)): + where = select.args.get("where") + if where is None: + continue + tables = _real_tables(select, ctes) + if not tables: + continue + for cond in _conjuncts(where.this): + parsed = _condition(cond) + if parsed is None: + continue + column, subject, pairs, bare = parsed + if not _timeish(subject, column, [b for _, b in pairs]): + continue + if column.table: + table = tables.get(name_key(column.table)) + elif len(tables) == 1: + table = next(iter(tables.values())) + else: + owners = [t for t in tables.values() + if name_key(column.name) in (_schema_columns(schema, t) or set())] + table = owners[0] if len(owners) == 1 else None + if table is None: + continue + shown = ".".join(p for p in (table.catalog, table.db, table.name) if p) + group = (index, name_key(shown), name_key(column.name)) + entry = found.setdefault(group, _Filter(group, table, shown, column, str(column.name))) + entry.conds.extend((op, bound, bare) for op, bound in pairs) + for entry in found.values(): + entry.key = (name_key(entry.table_name), name_key(entry.column_name), + tuple(sorted((op, bound.sql(), bare) for op, bound, bare in entry.conds))) + return list(found.values()) + + +# ========================================================================== +# 探查 +# ========================================================================== + + +async def _probe(source: Any, sql: str) -> tuple[list[Any], list[str | None]] | None: + """在数据源上跑一条探查查询：第一行的值、各列的类型（按位置）。走数据源查询同一道只读守卫；写语句、出错、超时都交回 + None。""" + from app.data.engine import run_query + from app.data.guard import QueryLimits, SqlRejected, check, is_write + + kind = (getattr(source, "kind", None) or "").lower() + try: + statement = check(sql, readonly=True, source_name=getattr(source, "name", ""), dialect=kind) + if is_write(statement, dialect=kind): + return None + result = await run_query(source, statement, limits=QueryLimits(timeout_seconds=PROBE_TIMEOUT_S, max_rows=2)) + except (SqlRejected, Exception): # noqa: BLE001 - 探查失败就当不知道 + logger.info("覆盖检查的探查查询没有成功（数据源 %s）", getattr(source, "name", ""), exc_info=True) + return None + if not result.rows: + return None + types = result.column_types or {} + return list(result.rows[0]), [types.get(c) for c in result.columns] + + +def _select_sql(items: list[str], read: str | None) -> str: + return f"SELECT {', '.join(items)}" + (" FROM DUAL" if read == "oracle" else "") + + +@dataclass +class Cache: + """同一个数据源上探查过的：表里那一列的范围、按运行当天现算的界。助手自查一轮里查好几次，不重复查库。""" + + ranges: dict[tuple[str, str, str], Any] = field(default_factory=dict) + bounds: dict[tuple[str, str], Any] = field(default_factory=dict) + + +@dataclass +class _Range: + """这一列在这张表里的范围。low / high 为 None 的是不知道那一头；empty 是表里这一列一个值都没有（现查出来的）。""" + + low: _Day | None + high: _Day | None + basis: str + day: str | None = None + empty: bool = False + + +#: 剖析事实（catalog_profile_jobs.last_facts 的一项）：表名 → 那一份，取不到为 None +Facts = Callable[[str], Awaitable[Mapping[str, Any] | None]] + + +async def _range(source: Any, flt: _Filter, read: str | None, cache: Cache, facts: Facts | None) -> _Range | None: + """这一列在这张表里的范围：现查 MIN / MAX（同一个数据源、同一张表、同一列查一次），查不成退回剖析、取值观察（facts）。""" + from sqlglot import exp + + key = (str(getattr(source, "name", "")), name_key(flt.table_name), name_key(flt.column_name)) + if key in cache.ranges: + return cache.ranges[key] + table = flt.table.copy() + table.set("alias", None) + ident = flt.column.this.copy() if isinstance(flt.column.this, exp.Identifier) \ + else exp.to_identifier(flt.column_name) + column = exp.Column(this=ident) + sql = (f"SELECT {exp.Min(this=column.copy()).sql(dialect=read)} AS lo, " + f"{exp.Max(this=column.copy()).sql(dialect=read)} AS hi FROM {table.sql(dialect=read)}") + found = await _probe(source, sql) + out: _Range | None = None + if found is not None: + values, kinds = found + low_raw, high_raw = (values + [None, None])[:2] + kinds = (kinds + [None, None])[:2] + if low_raw is None and high_raw is None: + out = _Range(None, None, "live", empty=True) + else: + low, high = _as_day(low_raw, kinds[0]), _as_day(high_raw, kinds[1]) + out = _Range(low, high, "live") if low is not None and high is not None else None + elif facts is not None: + out = await _fallback(flt, facts) + cache.ranges[key] = out + return out + + +async def _fallback(flt: _Filter, facts: Facts) -> _Range | None: + """现查不成：退回剖析记下的范围（date_ranges），再不然取值观察的最新日期（freshness，只知道最晚那一头）。""" + try: + found = await facts(flt.table_name) + except Exception: # noqa: BLE001 + return None + if not isinstance(found, Mapping): + return None + wanted = name_key(flt.column_name) + for item in found.get("date_ranges") or []: + if isinstance(item, Mapping) and name_key(str(item.get("column") or "")) == wanted: + lo, hi = _as_day(item.get("min")), _as_day(item.get("max")) + if lo is not None and hi is not None: + return _Range(lo, hi, "profile", day=str(found.get("at") or "")[:10] or None) + observed = found.get("observed") if isinstance(found.get("observed"), Mapping) else {} + for item in observed.get("freshness") or []: + if isinstance(item, Mapping) and name_key(str(item.get("column") or "")) == wanted: + hi = _as_day(item.get("max")) + if hi is not None: + return _Range(None, hi, "observed", day=str(item.get("at") or "")[:10] or None) + return None + + +def _literal_day(node: Any) -> _Day | None: + from sqlglot import exp + + from app.data.sqlcheck import _unparen + + node = _unparen(node) + negative = False + if isinstance(node, exp.Neg): + node, negative = _unparen(node.this), True + if not isinstance(node, exp.Literal) or negative: + return None + if node.is_string: + return _as_day(str(node.this)) + try: + value = int(str(node.this)) + except ValueError: + return None + return _as_day(value) + + +def _is_literal(node: Any) -> bool: + from sqlglot import exp + + from app.data.sqlcheck import _unparen + + node = _unparen(node) + if isinstance(node, exp.Neg): + node = _unparen(node.this) + return isinstance(node, exp.Literal) + + +async def _bounds(source: Any, nodes: list[Any], read: str | None, cache: Cache) -> dict[int, _Day | None]: + """各个界归一成日期：字面量直接解析，别的（按运行当天现算的表达式）一条 SELECT 在库里算出来。{id(节点): 日期}。""" + out: dict[int, _Day | None] = {} + pending: dict[str, list[int]] = {} + name = str(getattr(source, "name", "")) + for node in nodes: + if _is_literal(node): + out[id(node)] = _literal_day(node) + continue + try: + text = node.sql(dialect=read) + except Exception: # noqa: BLE001 + out[id(node)] = None + continue + if (name, text) in cache.bounds: + out[id(node)] = cache.bounds[(name, text)] + else: + pending.setdefault(text, []).append(id(node)) + texts = list(pending)[:MAX_BOUNDS] + found = await _probe(source, _select_sql([f"{t} AS b{i}" for i, t in enumerate(texts)], read)) if texts else None + for i, text in enumerate(pending): + day = None + if found is not None and i < len(texts): + values, kinds = found + day = _as_day(values[i], kinds[i]) if i < len(values) and i < len(kinds) else None + if i < len(texts): + cache.bounds[(name, text)] = day + for ident in pending[text]: + out[ident] = day + return out + + +@dataclass(frozen=True) +class Window: + """一段时间条件的范围（按天，两头都含）。start / end 为 None 的是那一头没有限制。""" + + start: date | None + end: date | None + + +def _window(conds: list[tuple[str, Any, bool]], days: Mapping[int, _Day | None], data: _Range, + read: str | None) -> Window | None: + """几个条件合成一段：下界取最晚的、上界取最早的；IN 列表取最早到最晚（中间没列到的日子也算进去，宁可判成有重叠）。 + 有一个界认不出、或者和列比不起来，整段不判（None）。「>」「<」的界不带时刻时按天挪一天（> 2026-10-11 从 10-12 算起）； + 带时刻的不挪（那一天里还可能有数据，宁可判成有重叠）。""" + lows: list[date] = [] + highs: list[date] = [] + points: list[date] = [] + sample = data.high or data.low + for op, node, bare in conds: + day = days.get(id(node)) + if day is None or (sample is not None and not _comparable(sample, day, bare=bare, read=read)): + return None + if op == "in": + points.append(day.day) + continue + if op in (">=", "="): + lows.append(day.day) + elif op == ">": + lows.append(day.day if day.timed else day.day + timedelta(days=1)) + if op in ("<=", "="): + highs.append(day.day) + elif op == "<": + highs.append(day.day if day.timed else day.day - timedelta(days=1)) + if points: + lows.append(min(points)) + highs.append(max(points)) + start = max(lows) if lows else None + end = min(highs) if highs else None + if start is None and end is None: + return None + if start is not None and end is not None and start > end: + return None + return Window(start, end) + + +def _disjoint(window: Window, data: _Range) -> bool: + """区间和表里的范围完全不重叠（表里这一列一个值都没有也算）。""" + if data.empty: + return True + if window.end is not None and data.low is not None and window.end < data.low.day: + return True + return window.start is not None and data.high is not None and window.start > data.high.day + + +# ========================================================================== +# 说法 +# ========================================================================== + + +def _window_text(window: Window) -> str: + if window.start is not None and window.end is not None: + return window.start.isoformat() if window.start == window.end \ + else f"{window.start.isoformat()} 到 {window.end.isoformat()}" + if window.start is not None: + return f"{window.start.isoformat()} 起" + return f"{window.end.isoformat()} 及以前" # type: ignore[union-attr] + + +def _data_text(data: _Range) -> str: + if data.empty: + return "一个值都没有" + if data.low is not None and data.high is not None: + if data.low.day == data.high.day: + return f"只有 {data.low.day.isoformat()} 这一天的数据" + return f"只有 {data.low.day.isoformat()} 到 {data.high.day.isoformat()} 的数据" + return f"最新只到 {data.high.day.isoformat()}" # type: ignore[union-attr] + + +def _basis_text(data: _Range) -> str: + if data.basis == "profile": + return BASIS_PROFILE.format(day=data.day or "上次") + if data.basis == "observed": + return BASIS_OBSERVED.format(day=data.day or "上次") + return "" + + +def _columns_text(columns: list[str]) -> str: + head = f"「{columns[0]}」" + return f"{head}等 {len(columns)} 列" if len(columns) > 1 else head + + +def _range_dict(data: _Range) -> list[str | None] | None: + if data.empty: + return None + return [data.low.day.isoformat() if data.low else None, data.high.day.isoformat() if data.high else None] + + +def _window_dict(window: Window) -> list[str | None]: + return [window.start.isoformat() if window.start else None, window.end.isoformat() if window.end else None] + + +# ========================================================================== +# 入口 +# ========================================================================== + + +def zero_columns(payload: Mapping[str, Any]) -> list[str]: + """查询结果里「整列都是 0 或空」的数值列（按 column_types 是数的、或者没记类型却全是数和空值的）。一行都没有的结果 + 返回空列表：那归「没有查到」管，不归这里。""" + from app.engine.expressions import numeric_text + + columns = [str(c) for c in payload.get("columns") or []] + rows = payload.get("rows") or [] + if not columns or not isinstance(rows, list) or not rows: + return [] + types = payload.get("column_types") if isinstance(payload.get("column_types"), Mapping) else {} + out: list[str] = [] + for index, column in enumerate(columns): + if types.get(column) not in (None, "number"): + continue + values = [row[index] if isinstance(row, (list, tuple)) and index < len(row) + else (row.get(column) if isinstance(row, Mapping) else None) for row in rows] + + def zero(value: Any) -> bool: + if value is None: + return True + if isinstance(value, bool): + return False + if isinstance(value, (int, float)): + return value == 0 + number = numeric_text(value) if isinstance(value, str) else None + return number is not None and number == 0 + + if all(zero(v) for v in values): + out.append(column) + return out + + +async def _judge(source: Any, flts: list[_Filter], read: str | None, cache: Cache, facts: Facts | None, + masked: set[str], deadline: float) -> dict[tuple[int, str, str], tuple[str, Window | None, _Range | None]]: + """每段时间条件判一下：none（区间里表里没有数据）/ some（有重叠）/ unknown（判断不了），带上区间和范围。""" + out: dict[tuple[int, str, str], tuple[str, Window | None, _Range | None]] = {} + ranges: dict[tuple[str, str], _Range | None] = {} + usable = [f for f in flts if name_key(f.column_name) not in masked] + for flt in flts: + if flt not in usable: + out[flt.key] = ("unknown", None, None) + for flt in usable: + if time.monotonic() > deadline: + out[flt.key] = ("unknown", None, None) + continue + rkey = (name_key(flt.table_name), name_key(flt.column_name)) + if rkey not in ranges: + ranges[rkey] = await _range(source, flt, read, cache, facts) if len(ranges) < MAX_RANGES else None + data = ranges[rkey] + if data is None: + out[flt.key] = ("unknown", None, None) + continue + days = await _bounds(source, [node for _, node, _ in flt.conds], read, cache) + window = _window(flt.conds, days, data, read) + if window is None: + out[flt.key] = ("unknown", None, data) + else: + out[flt.key] = ("none" if _disjoint(window, data) else "some", window, data) + return out + + +async def result_gaps(source: Any, sql: str, payload: Mapping[str, Any], *, facts: Facts | None = None, + masked: Iterable[str] = (), cache: Cache | None = None, + budget: float = BUDGET_S) -> list[dict[str, Any]]: + """一份查询结果里，整列是 0 或空、而查询区间里这张表没有数据的那几列：[{columns, table, column, window, data_range, + basis, basis_at?, message}]，同一张表、同一列、同一段区间的几列并成一条。没有问题（或者判断不了）返回空列表。从不抛异常。""" + try: + return await asyncio.wait_for(_result_gaps(source, sql, payload, facts, set(masked), cache or Cache(), budget), + timeout=budget + PROBE_TIMEOUT_S) + except asyncio.TimeoutError: + logger.warning("覆盖检查超过时限，本次跳过（数据源 %s）", getattr(source, "name", "")) + except Exception: # noqa: BLE001 - 检查不能让一次成功的查询变成失败 + logger.exception("覆盖检查出错，本次跳过（数据源 %s）", getattr(source, "name", "")) + return [] + + +async def _result_gaps(source: Any, sql: str, payload: Mapping[str, Any], facts: Facts | None, masked: set[str], + cache: Cache, budget: float) -> list[dict[str, Any]]: + zeros = zero_columns(payload) + if not zeros: + return [] + read = _read(getattr(source, "kind", None)) + tree = _parse(sql, read) + if tree is None: + return [] + masked = {m.strip().lower() for m in masked if isinstance(m, str) and m.strip()} + schema = getattr(source, "schema_cache", None) + per_column = {c: filters(narrowed(tree, c, read), schema) for c in zeros if c.lower() not in masked} + every = {f.key: f for flts in per_column.values() for f in flts} + if not every: + return [] + judged = await _judge(source, list(every.values()), read, cache, facts, masked, time.monotonic() + budget) + groups: dict[tuple[Any, ...], dict[str, Any]] = {} + for column, flts in per_column.items(): + verdicts = [judged.get(f.key) for f in flts] + if not flts or any(v is None or v[0] != "none" for v in verdicts): + continue + for flt, (_, window, data) in zip(flts, verdicts): + gkey = (name_key(flt.table_name), name_key(flt.column_name), window, data.basis if data else None) + group = groups.setdefault(gkey, {"flt": flt, "window": window, "data": data, "columns": []}) + if column not in group["columns"]: + group["columns"].append(column) + out = [] + for group in groups.values(): + flt, window, data, columns = group["flt"], group["window"], group["data"], group["columns"] + out.append({ + "columns": columns[:MAX_COLUMNS], "table": flt.table_name, "column": flt.column_name, + "window": _window_dict(window), "data_range": _range_dict(data), "basis": data.basis, + **({"basis_at": data.day} if data.day else {}), + "message": NO_DATA_MESSAGE.format(columns=_columns_text(columns), window=_window_text(window), + table=flt.table_name, column=flt.column_name, + data=_data_text(data), basis=_basis_text(data)), + }) + return out + + +async def sql_gaps(source: Any, sql: str, *, facts: Facts | None = None, masked: Iterable[str] = (), + cache: Cache | None = None, budget: float = BUDGET_S) -> list[dict[str, Any]]: + """还没执行的一条 SQL（助手自查）：SQL 里每一段时间条件，区间和表里的范围完全不重叠的那几段：[{table, column, + window, data_range, basis, basis_at?, message}]，同一张表、同一列、同一段区间只算一条。从不抛异常。""" + try: + return await asyncio.wait_for(_sql_gaps(source, sql, facts, set(masked), cache or Cache(), budget), + timeout=budget + PROBE_TIMEOUT_S) + except asyncio.TimeoutError: + logger.warning("自查的覆盖检查超过时限，本次跳过（数据源 %s）", getattr(source, "name", "")) + except Exception: # noqa: BLE001 + logger.exception("自查的覆盖检查出错，本次跳过（数据源 %s）", getattr(source, "name", "")) + return [] + + +async def _sql_gaps(source: Any, sql: str, facts: Facts | None, masked: set[str], cache: Cache, + budget: float) -> list[dict[str, Any]]: + read = _read(getattr(source, "kind", None)) + tree = _parse(sql, read) + if tree is None: + return [] + masked = {m.strip().lower() for m in masked if isinstance(m, str) and m.strip()} + flts = filters(tree, getattr(source, "schema_cache", None)) + if not flts: + return [] + judged = await _judge(source, flts, read, cache, facts, masked, time.monotonic() + budget) + out: list[dict[str, Any]] = [] + seen: set[tuple[Any, ...]] = set() + for flt in flts: + verdict, window, data = judged.get(flt.key, ("unknown", None, None)) + key = (name_key(flt.table_name), name_key(flt.column_name), window) + if verdict != "none" or window is None or data is None or key in seen: + continue + seen.add(key) + out.append({"table": flt.table_name, "column": flt.column_name, "window": _window_dict(window), + "data_range": _range_dict(data), "basis": data.basis, **({"basis_at": data.day} if data.day else {}), + "message": NO_DATA_SQL_MESSAGE.format(window=_window_text(window), table=flt.table_name, + column=flt.column_name, data=_data_text(data), + basis=_basis_text(data))}) + return out + + +# ========================================================================== +# 读快照里记下的 +# ========================================================================== + + +def no_data_of(snapshot: Any, column: Any) -> dict[str, Any] | None: + """查询快照里记着 column 这一列「查询区间里这张表没有数据」的那一条（no_data）；没有为 None。按列名认（不分大小写、 + 不计空白）。""" + if not isinstance(snapshot, Mapping) or not isinstance(column, str): + return None + for item in snapshot.get("no_data") or []: + if isinstance(item, Mapping) and any(isinstance(c, str) and _same_name(c, column) + for c in item.get("columns") or []): + return dict(item) + return None + + +__all__ = ["BUDGET_S", "Cache", "NO_DATA_SHORT", "PROBE_TIMEOUT_S", "Window", "filters", "narrowed", "no_data_of", + "result_gaps", "sql_gaps", "zero_columns"] diff --git a/backend/app/tools/datasource.py b/backend/app/tools/datasource.py index fe249549..c54a6823 100644 --- a/backend/app/tools/datasource.py +++ b/backend/app/tools/datasource.py @@ -696,6 +696,12 @@ def _make_query_tool(source: Any, ctx: ToolContext, *, fixed: bool = False, # 快照和以前一字不差 if checks := await _sql_checks(_checker(), sql): payload["checks"] = checks + # 结果里有整列是 0 或空的数值列时，现查查询区间里这张表有没有数据（data/coverage.py）：没有就照实写进结果和 + # 快照（no_data，给人和模型看的 message），运行事件里记一条 query_no_data。没查出问题不加这个键，快照和以前一字不差 + if gaps := await _coverage(source, str(payload.get("sql") or sql), payload, mask): + payload["no_data"] = gaps + await _announce(ctx, EventType.LOG, level="warn", code="query_no_data", no_data=gaps, + message="；".join(g["message"] for g in gaps)) if outside and (unchecked := _unchecked_tables(sql, source, outside)): payload["unchecked_tables"] = unchecked # 表结构快照：这次查询时数据源的结构冻结下来，报告里写的表名、字段名按它核对。事后数据源 @@ -776,6 +782,33 @@ def _unchecked_tables(sql: str, source: Any, outside: frozenset[str]) -> list[st if name_key(str(t).rsplit(".", 1)[-1]) in outside and data_catalog.resolve_table_name(cache, str(t)) is None] +async def _coverage(source: Any, sql: str, payload: Mapping[str, Any], mask: list[str]) -> list[dict[str, Any]]: + """查询区间里这张表没有数据的那几列（data/coverage.result_gaps）。只在结果里有整列是 0 或空的数值列时才查库；遮罩列 + 不探查；现查表的范围不成时退回剖析、取值观察记下的范围（上传表格的快照对不上就不用）。从不抛异常。""" + from app.data import coverage + + if not coverage.zero_columns(payload): + return [] + + async def facts(table: str) -> Mapping[str, Any] | None: + from app.data.catalog_profile_jobs import last_facts + from app.db.base import SessionLocal + + source_id = getattr(source, "id", None) + if not source_id: + return None + name = data_catalog.resolve_table_name(getattr(source, "schema_cache", None), table) or table + async with SessionLocal() as session: + found = (await last_facts(session, str(source_id), [name])).get(name) + snapshot = getattr(source, "snapshot_id", None) + if found and snapshot and found.get("snapshot_id") != snapshot: + # 上传表格：剖析时是另一版数据，范围对不上这次读的这一版 + return None + return found + + return await coverage.result_gaps(source, sql, payload, facts=facts, masked=mask) + + async def _sql_checks(checker: SqlChecker, sql: str) -> list[dict[str, Any]]: """一条查询的 SQL 检查结果（as_dict 的形状）。在线程里跑、限时 SQL_CHECK_TIMEOUT_S：解析是纯计算，放在事件循环里 会挡住同一进程里别的运行。出错、超时都返回空列表——检查不能让一次成功的查询变成失败。""" diff --git a/backend/tests/test_query_coverage.py b/backend/tests/test_query_coverage.py new file mode 100644 index 00000000..b8300d4f --- /dev/null +++ b/backend/tests/test_query_coverage.py @@ -0,0 +1,234 @@ +"""查询区间里这张表没有数据时，查询结果里照实说（data/coverage.py，接在 tools/datasource 的查询工具上）。 + +部署上撞过：一张到访事件表只存最新一天（每天整表重灌），周报按运行当天现算的区间取上周、上上周的到访人数，6 列全是 0， +报告照写「上周到访人数为 0 人」。剖析记过日期范围，但对滚动表不可靠，所以运行时现查： + +- 结果里有整列是 0 或空的数值列时，把 SQL 收窄到这一列，认出它用到的表和时间谓词；界是字面量的直接解析，按运行当天现算 + 的在同一个库里算出来；这一列在表里的范围现查 MIN / MAX，查不成退回剖析记下的范围（照实标「按 X 日的剖析」）； +- 区间和范围完全不重叠才算没有数据：结果和快照里多一项 no_data（给人和模型看的 message），运行事件里一条 query_no_data； + 没查出问题不加这个键，快照和以前一字不差； +- 有数的列、一行都没有的结果、遮罩列、写法对不上（文本列 '20261012' 对文本界 '2026-10-12'）、认不出的一律不判。 + +库是现造的 SQLite 文件，表名、列名、数字都是虚构的。 +""" +from __future__ import annotations + +import json +import sqlite3 +import uuid + +import pytest + +from app.core import artifact_store +from app.data import coverage, introspect +from app.data.engine import engines +from app.db.base import SessionLocal +from app.db.models import DataSource +from app.tools.datasource import build_datasource_tools +from app.tools.registry import ToolContext, Versions + +#: 这张表只存一天（2020-01-01），按今天现算的上周一定落在它外面 +ONLY_DAY = "20200101" +#: 按今天现算的「上周」（SQLite 的写法，和列一样写成 'YYYYMMDD'） +WINDOW = "{c} >= strftime('%Y%m%d', 'now', '-14 days') AND {c} < strftime('%Y%m%d', 'now', '-7 days')" +KPI_SQL = ("SELECT (SELECT COUNT(DISTINCT v.member_id) FROM gate_events v WHERE " + WINDOW.format(c="v.event_date") + + ") AS 上周到访人数, (SELECT COUNT(*) FROM gate_events v WHERE " + WINDOW.format(c="v.event_date") + + ") AS 上周到访人次, (SELECT SUM(pv) FROM daily_pv WHERE day >= date('now', '-14 days')) AS 客流PV") + + +@pytest.fixture +async def plant(tmp_path): + """gate_events 只有 ONLY_DAY 一天；daily_pv 覆盖最近 30 天、每天 100。""" + path = tmp_path / f"{uuid.uuid4().hex[:8]}.db" + db = sqlite3.connect(path) + db.executescript( + "CREATE TABLE gate_events (event_date TEXT, member_id INTEGER, gate TEXT);" + f"INSERT INTO gate_events VALUES ('{ONLY_DAY}', 1, 'A'), ('{ONLY_DAY}', 2, 'B');" + "CREATE TABLE daily_pv (day TEXT, pv INTEGER);" + "CREATE TABLE iso_events (paid_on TEXT, amount INTEGER);" + "INSERT INTO iso_events VALUES ('2020-01-01', 5);") + for back in range(30): + db.execute("INSERT INTO daily_pv VALUES (date('now', ?), 100)", (f"-{back} days",)) + db.commit() + db.close() + name = f"plant_{uuid.uuid4().hex[:8]}" + async with SessionLocal() as session: + row = DataSource(name=name, kind="sqlite", database=str(path), readonly=True, enabled=True) + session.add(row) + await session.flush() + row.schema_cache = await introspect.introspect(row) + await session.commit() + source_id = row.id + yield name, source_id + await engines.invalidate(source_id) + + +async def _query(name: str, sql: str, events: list | None = None) -> dict: + async def emit(event_type, **data): + if events is not None: + events.append((str(event_type), data)) + + async with SessionLocal() as session: + built = await build_datasource_tools([f"db_query__{name}"], ToolContext( + data_versions=Versions.CURRENT, run_id="run-coverage", node_id="q_kpi", emit=emit), session) + got = await built[0].ainvoke({"sql": sql}) + return json.loads(got) + + +async def test_zero_columns_over_a_window_without_data_are_flagged(plant): + name, _ = plant + events: list = [] + result = await _query(name, KPI_SQL, events) + assert result["rows"][0][:2] == [0, 0] + [gap] = result["no_data"] + assert gap["columns"] == ["上周到访人数", "上周到访人次"] and gap["table"] == "gate_events" + assert gap["column"] == "event_date" and gap["data_range"] == ["2020-01-01", "2020-01-01"] and gap["basis"] == "live" + start, end = gap["window"] + assert start < end and start > "2020-01-01" + assert gap["message"].startswith("「上周到访人数」等 2 列查的是 ") and "只有 2020-01-01 这一天的数据" in gap["message"] + assert "这些 0 不代表这段时间真的是 0，是这段时间表里没有数据" in gap["message"] + # 有数的列（客流 PV）不受影响 + assert all("客流PV" not in g["columns"] for g in result["no_data"]) + # 快照里也有；运行事件里有一条 query_no_data + snapshot = artifact_store.load(result["artifact"]) + assert snapshot["no_data"] == result["no_data"] + [(etype, data)] = [e for e in events if e[1].get("code") == "query_no_data"] + assert data["level"] == "warn" and data["no_data"] == result["no_data"] + + +async def test_literal_windows_are_parsed_directly(plant): + name, _ = plant + sql = ("SELECT COUNT(*) AS 到访人次 FROM gate_events WHERE event_date >= '20261012' AND event_date < '20261019'") + [gap] = (await _query(name, sql))["no_data"] + assert gap["window"] == ["2026-10-12", "2026-10-18"] # < 不带时刻：截到前一天 + + +async def test_nothing_is_added_when_the_window_has_data_or_the_column_is_not_zero(plant): + name, _ = plant + for sql in ( + f"SELECT COUNT(*) AS n FROM gate_events WHERE event_date = '{ONLY_DAY}'", # 有数 + "SELECT SUM(pv) AS pv FROM daily_pv WHERE day >= date('now', '-7 days')", # 有数 + "SELECT COUNT(*) AS n FROM gate_events WHERE event_date >= '20200101'", # 区间覆盖了那一天 + "SELECT gate, COUNT(*) AS n FROM gate_events WHERE event_date >= '20261012' GROUP BY gate", # 一行都没有 + ): + result = await _query(name, sql) + assert "no_data" not in result, (sql, result.get("no_data")) + assert "no_data" not in artifact_store.load(result["artifact"]), "没查出问题：快照和以前一字不差" + + +async def test_a_text_column_against_a_differently_written_text_bound_is_not_judged(plant): + """列里存的是 '20200101'，界写的是 '2026-10-12'：库按字符串比，和按日期比对不上，不判。""" + name, _ = plant + result = await _query(name, "SELECT COUNT(*) AS n FROM gate_events WHERE event_date >= '2026-10-12'") + assert "no_data" not in result + # ISO 写法对 ISO 写法：照常判 + result = await _query(name, "SELECT COUNT(*) AS n FROM iso_events WHERE paid_on >= date('now', '-7 days')") + [gap] = result["no_data"] + assert gap["column"] == "paid_on" and gap["data_range"] == ["2020-01-01", "2020-01-01"] + + +async def test_masked_columns_are_not_probed(plant): + name, source_id = plant + async with SessionLocal() as session: + row = await session.get(DataSource, source_id) + row.options = {**(row.options or {}), "mask_columns": ["event_date"]} + await session.commit() + result = await _query(name, "SELECT COUNT(*) AS n FROM gate_events WHERE event_date >= '20261012'") + assert "no_data" not in result + + +async def test_a_failed_probe_falls_back_to_the_profile(plant, monkeypatch): + """现查不成（超时、出错）：退回剖析记下的范围，照实标「按 X 日的剖析」。""" + name, _ = plant + real = coverage._probe + + async def probe(source, sql): + return None if "MIN(" in sql else await real(source, sql) + + async def facts(table): + assert table == "gate_events" + return {"at": "2026-10-01T03:00:00+00:00", "date_ranges": [{"column": "event_date", "min": "20260930", + "max": "20260930"}]} + + monkeypatch.setattr(coverage, "_probe", probe) + async with SessionLocal() as session: + source = await session.get(DataSource, (await _source_id(name))) + payload = {"columns": ["n"], "rows": [[0]], "row_count": 1, "column_types": {"n": "number"}} + [gap] = await coverage.result_gaps(source, "SELECT COUNT(*) AS n FROM gate_events WHERE event_date >= '20261012'", + payload, facts=facts) + assert gap["basis"] == "profile" and gap["basis_at"] == "2026-10-01" + assert gap["data_range"] == ["2026-09-30", "2026-09-30"] and "（现查表的范围没有成功，按 2026-10-01 的剖析）" in gap["message"] + # 剖析也没有：不判 + assert await coverage.result_gaps(source, "SELECT COUNT(*) AS n FROM gate_events WHERE event_date >= '20261012'", + payload, facts=None) == [] + + +async def _source_id(name: str) -> str: + from sqlalchemy import select + + async with SessionLocal() as session: + return (await session.execute(select(DataSource.id).where(DataSource.name == name))).scalar_one() + + +@pytest.mark.parametrize("value, kind, day, shape", [ + ("2026-10-12", None, "2026-10-12", "iso"), + ("2026-10-12T08:30:00", "datetime", "2026-10-12", "date"), + ("20261012", None, "2026-10-12", "compact"), + (20261012, None, "2026-10-12", "number"), + ("2026-13-01", None, None, None), + ("2026/10/12", None, None, None), + ("上周", None, None, None), +]) +def test_dates_are_normalised(value, kind, day, shape): + got = coverage._as_day(value, kind) + assert (got.day.isoformat() if got else None) == day and (got.shape if got else None) == shape + + +def test_zero_columns(): + payload = {"columns": ["a", "b", "c", "d"], "column_types": {"a": "number", "b": "number", "d": "text"}, + "rows": [[0, 1, None, "0"], ["0.00", 0, None, "0"]]} + assert coverage.zero_columns(payload) == ["a", "c"] + assert coverage.zero_columns({**payload, "rows": []}) == [] + + +async def test_the_weekly_report_shape_on_mysql(monkeypatch): + """部署上那张周报的形状（MySQL）：一条查询用标量子查询一次取几列，上周、上上周按运行当天现算；到访表只有最新一天， + 列存成 VARCHAR 'YYYYMMDD'。界由库算出来是 DATE 类型：MySQL 把文本列转成日期再比，照常判。两段区间各记一条。""" + from types import SimpleNamespace + + last = ("event_date >= DATE_SUB(CURRENT_DATE, INTERVAL (WEEKDAY(CURRENT_DATE) + 7) DAY) " + "AND event_date < DATE_SUB(CURRENT_DATE, INTERVAL WEEKDAY(CURRENT_DATE) DAY)") + before = ("event_date >= DATE_SUB(CURRENT_DATE, INTERVAL (WEEKDAY(CURRENT_DATE) + 14) DAY) " + "AND event_date < DATE_SUB(CURRENT_DATE, INTERVAL (WEEKDAY(CURRENT_DATE) + 7) DAY)") + sql = (f"SELECT (SELECT SUM(pv) FROM pv_daily WHERE stat_day >= '2026-10-01') AS `客流PV`, " + f"(SELECT COUNT(DISTINCT member_id) FROM gate_events WHERE {last}) AS `上周到访人数`, " + f"(SELECT COUNT(*) FROM gate_events WHERE {last}) AS `上周到访人次`, " + f"(SELECT COUNT(DISTINCT member_id) FROM gate_events WHERE {before}) AS `上上周到访人数`") + days = {"(WEEKDAY(CURRENT_DATE) + 14)": "2026-09-28", "(WEEKDAY(CURRENT_DATE) + 7)": "2026-10-05", + "(WEEKDAY(CURRENT_DATE)) DAY": "2026-10-12"} # sqlglot 写回 MySQL 时给这一项加了括号 + seen: list[str] = [] + + async def probe(source, text): + seen.append(text) + if "MIN(" in text: + return ["20261019", "20261019"], ["text", "text"] + values = [] + for part in text[len("SELECT "):].split(" AS b")[:-1]: + values.append(next(v for k, v in days.items() if k in part)) + return values, ["date"] * len(values) + + monkeypatch.setattr(coverage, "_probe", probe) + source = SimpleNamespace(name="plant", kind="mysql", schema_cache=None, id=None) + payload = {"columns": ["客流PV", "上周到访人数", "上周到访人次", "上上周到访人数"], + "column_types": {"客流PV": "number", "上周到访人数": "number", "上周到访人次": "number", + "上上周到访人数": "number"}, + "rows": [[3200, 0, 0, 0]], "row_count": 1} + gaps = await coverage.result_gaps(source, sql, payload) + by_window = {tuple(g["window"]): g for g in gaps} + assert set(by_window) == {("2026-10-05", "2026-10-11"), ("2026-09-28", "2026-10-04")} + assert by_window[("2026-10-05", "2026-10-11")]["columns"] == ["上周到访人数", "上周到访人次"] + assert by_window[("2026-09-28", "2026-10-04")]["columns"] == ["上上周到访人数"] + assert all(g["data_range"] == ["2026-10-19", "2026-10-19"] for g in gaps) + # 表的范围同一张表、同一列只查一次；按运行当天现算的界一条 SELECT 算出来 + assert sum("MIN(" in t for t in seen) == 1 + assert all(t.startswith("SELECT ") for t in seen)