From 680ea4ecf83dcbbefbf2b4147e7155b6a203618b Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Tue, 20 Oct 2026 10:58:20 +0800 Subject: [PATCH] =?UTF-8?q?=E8=A6=86=E7=9B=96=E6=A3=80=E6=9F=A5=EF=BC=9A?= =?UTF-8?q?=E7=8E=B0=E6=9F=A5=E6=B2=A1=E6=88=90=E5=8A=9F=EF=BC=88=E5=A4=A7?= =?UTF-8?q?=E8=A1=A8=E5=85=A8=E8=A1=A8=20MIN=20/=20MAX=20=E8=B6=85?= =?UTF-8?q?=E6=97=B6=E3=80=81=E5=BA=93=E4=B8=80=E6=97=B6=E8=BF=9E=E4=B8=8D?= =?UTF-8?q?=E4=B8=8A=EF=BC=89=E4=B9=9F=E8=AE=B0=20LIVE=5FTTL=5FS=EF=BC=8C?= =?UTF-8?q?=E6=9C=9F=E9=97=B4=E5=90=8C=E4=B8=80=E5=BC=A0=E8=A1=A8=E7=9B=B4?= =?UTF-8?q?=E6=8E=A5=E9=80=80=E5=9B=9E=E5=89=96=E6=9E=90=E3=80=81=E5=8F=96?= =?UTF-8?q?=E5=80=BC=E8=A7=82=E5=AF=9F=EF=BC=8C=E4=B8=8D=E5=86=8D=E6=AF=8F?= =?UTF-8?q?=E6=AC=A1=E7=99=BD=E7=AD=89=20PROBE=5FTIMEOUT=5FS=20=E7=A7=92?= =?UTF-8?q?=EF=BC=88=E6=B5=8B=E8=AF=95=20test=5Fa=5Ffailed=5Fprobe=5Fis=5F?= =?UTF-8?q?not=5Fretried=5Ffor=5Fa=5Fwhile=EF=BC=89=EF=BC=9B=E6=96=B9?= =?UTF-8?q?=E8=A8=80=E5=AE=B9=E5=99=A8=E9=9B=86=E6=88=90=E6=B5=8B=E8=AF=95?= =?UTF-8?q?=EF=BC=88scripts/dialect=5Fit.py=EF=BC=89=E5=8A=A0=E4=B8=80?= =?UTF-8?q?=E6=AE=B5=E8=A6=86=E7=9B=96=E6=A3=80=E6=9F=A5=EF=BC=9A=E5=8F=AA?= =?UTF-8?q?=E5=AD=98=E5=BD=93=E5=A4=A9=E7=9A=84=E6=BB=9A=E5=8A=A8=E8=A1=A8?= =?UTF-8?q?=EF=BC=88DATE=20=E5=88=97=EF=BC=8CMySQL=20/=20MariaDB=20?= =?UTF-8?q?=E5=8F=A6=E6=9C=89=E6=8C=89=E6=96=87=E5=AD=97=20'YYYYMMDD'=20?= =?UTF-8?q?=E5=AD=98=E7=9A=84=EF=BC=89=EF=BC=8C=E6=8C=89=E8=BF=90=E8=A1=8C?= =?UTF-8?q?=E5=BD=93=E5=A4=A9=E7=8E=B0=E7=AE=97=E7=9A=84=E4=B8=8A=E5=91=A8?= =?UTF-8?q?=E5=8C=BA=E9=97=B4=E5=88=A4=E6=88=90=E6=B2=A1=E6=9C=89=E6=95=B0?= =?UTF-8?q?=E6=8D=AE=E3=80=81=E5=8C=BA=E9=97=B4=E5=8C=85=E5=90=AB=E5=BD=93?= =?UTF-8?q?=E5=A4=A9=E5=8F=AA=E6=98=AF=E5=88=AB=E7=9A=84=E6=9D=A1=E4=BB=B6?= =?UTF-8?q?=E7=AD=9B=E7=A9=BA=E7=9A=84=E4=B8=8D=E6=8A=A5=EF=BC=8C=E6=9F=A5?= =?UTF-8?q?=E8=AF=A2=E5=90=8E=E5=92=8C=E5=8A=A9=E6=89=8B=E8=87=AA=E6=9F=A5?= =?UTF-8?q?=E4=B8=A4=E4=B8=AA=E5=85=A5=E5=8F=A3=E9=83=BD=E6=A0=B8=E2=80=94?= =?UTF-8?q?=E2=80=94PostgreSQL=2016=E3=80=81MySQL=208.4=E3=80=81MariaDB=20?= =?UTF-8?q?11=E3=80=81Oracle=20Free=2023=20=E5=85=A8=E8=BF=87?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/app/data/coverage.py | 26 +++++++++++---- backend/tests/test_query_coverage.py | 29 ++++++++++++++++ scripts/dialect_it.py | 50 +++++++++++++++++++++++++++- 3 files changed, 97 insertions(+), 8 deletions(-) diff --git a/backend/app/data/coverage.py b/backend/app/data/coverage.py index 6e2960d9..e7db5445 100644 --- a/backend/app/data/coverage.py +++ b/backend/app/data/coverage.py @@ -52,6 +52,9 @@ MAX_COLUMNS = 12 #: MAX。滚动表一天才换一次数据，几分钟的旧不影响判断；查不成的、退回剖析的不记 LIVE_TTL_S = 300.0 _LIVE: dict[tuple[Any, ...], tuple[float, Any]] = {} +#: 现查没成功（大表全表 MIN / MAX 超时、库一时连不上）也记 LIVE_TTL_S：这段时间里同一张表直接退回剖析、取值观察， +#: 不然取数 agent 连查几次、每次结果里有整列 0，每次都要白等 PROBE_TIMEOUT_S 秒才退回 +_PROBE_FAILED = object() #: 给人和模型看的说法 NO_DATA_MESSAGE = "{columns}查的是 {window}，但 {table} 的 {column} {data}：这些 0 不代表这段时间真的是 0，是这段时间表里没有数据{basis}" @@ -398,9 +401,12 @@ async def _range(source: Any, flt: _Filter, read: str | None, cache: Cache, fact return cache.ranges[key] live = (getattr(source, "id", None), getattr(source, "snapshot_id", None), *key) held = _LIVE.get(live) + failed_before = False if held is not None and held[0] > time.monotonic(): - cache.ranges[key] = held[1] - return held[1] + if held[1] is not _PROBE_FAILED: + cache.ranges[key] = held[1] + return held[1] + failed_before = True table = flt.table.copy() table.set("alias", None) ident = flt.column.this.copy() if isinstance(flt.column.this, exp.Identifier) \ @@ -408,7 +414,7 @@ async def _range(source: Any, flt: _Filter, read: str | None, cache: Cache, fact 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) + found = None if failed_before else await _probe(source, sql) out: _Range | None = None if found is not None: values, kinds = found @@ -422,13 +428,19 @@ async def _range(source: Any, flt: _Filter, read: str | None, cache: Cache, fact elif facts is not None: out = await _fallback(flt, facts) cache.ranges[key] = out - if out is not None and out.basis == "live": - if len(_LIVE) > 512: - _LIVE.clear() - _LIVE[live] = (time.monotonic() + LIVE_TTL_S, out) + if found is None and not failed_before: + _remember(live, _PROBE_FAILED) + elif out is not None and out.basis == "live": + _remember(live, out) return out +def _remember(live: tuple[Any, ...], value: Any) -> None: + if len(_LIVE) > 512: + _LIVE.clear() + _LIVE[live] = (time.monotonic() + LIVE_TTL_S, value) + + async def _fallback(flt: _Filter, facts: Facts) -> _Range | None: """现查不成：退回剖析记下的范围（date_ranges），再不然取值观察的最新日期（freshness，只知道最晚那一头）。""" try: diff --git a/backend/tests/test_query_coverage.py b/backend/tests/test_query_coverage.py index c8ce0941..f8fcfe54 100644 --- a/backend/tests/test_query_coverage.py +++ b/backend/tests/test_query_coverage.py @@ -257,3 +257,32 @@ async def test_a_live_range_is_kept_for_a_while(monkeypatch): await coverage.result_gaps(source, sql, payload) await coverage.result_gaps(source, sql, payload) assert sum("MIN(" in t for t in seen) == 3, "过期了就重新现查" + + +async def test_a_failed_probe_is_not_retried_for_a_while(monkeypatch): + """现查没成功（大表全表 MIN / MAX 超时）也记几分钟：期间同一张表直接退回剖析，不再每次白等探查的时限。""" + from types import SimpleNamespace + + seen: list[str] = [] + + async def probe(source, text): + seen.append(text) + return None + + async def facts(table): + return {"at": "2026-10-01T03:00:00+00:00", "date_ranges": [{"column": "event_date", "min": "20260930", + "max": "20260930"}]} + + monkeypatch.setattr(coverage, "_probe", probe) + monkeypatch.setattr(coverage, "_LIVE", {}) + source = SimpleNamespace(name=f"plant_{uuid.uuid4().hex[:6]}", kind="sqlite", schema_cache=None, id="s-failed") + payload = {"columns": ["n"], "rows": [[0]], "row_count": 1, "column_types": {"n": "number"}} + sql = "SELECT COUNT(*) AS n FROM gate_events WHERE event_date >= '20261012'" + for _ in range(3): + [gap] = await coverage.result_gaps(source, sql, payload, facts=facts) + assert gap["basis"] == "profile" + assert sum("MIN(" in t for t in seen) == 1, "失败的现查也记着，期间不再白等" + monkeypatch.setattr(coverage, "LIVE_TTL_S", 0.0) + coverage._LIVE.clear() + await coverage.result_gaps(source, sql, payload, facts=facts) + assert sum("MIN(" in t for t in seen) == 2, "过期了就重新现查" diff --git a/scripts/dialect_it.py b/scripts/dialect_it.py index f5621d42..362df5a6 100644 --- a/scripts/dialect_it.py +++ b/scripts/dialect_it.py @@ -149,6 +149,23 @@ def _ddl(t: Target) -> list[str]: return out +def _snapshot(t: Target) -> list[str]: + """覆盖检查（data/coverage.py）的夹具：只存当天的滚动表（每天整表重灌的那种）。日期按库自己的当前日期写，避开容器 + 和本机时区不一致；daily_snapshot 的日期按文字 'YYYYMMDD' 存（部署上撞过的写法，只在 MySQL / MariaDB 上判）， + daily_snapshot_d 按 DATE 存。""" + vc = "VARCHAR2" if t.oracle else "VARCHAR" + today_text = ("TO_CHAR(SYSDATE, 'YYYYMMDD')" if t.oracle + else "REPLACE(CAST(CURRENT_DATE AS CHAR), '-', '')" if t.mysql + else "REPLACE(CAST(CURRENT_DATE AS VARCHAR), '-', '')") + today = "TRUNC(SYSDATE)" if t.oracle else "CURRENT_DATE" + out = [f"CREATE TABLE daily_snapshot (event_day {vc}(8), visitor {vc}(16))", + f"CREATE TABLE daily_snapshot_d (event_day DATE, visitor {vc}(16))"] + for who in ("v1", "v2", "v3"): + out.append(f"INSERT INTO daily_snapshot (event_day, visitor) VALUES ({today_text}, '{who}')") + out.append(f"INSERT INTO daily_snapshot_d (event_day, visitor) VALUES ({today}, '{who}')") + return out + + def _paid(t: Target, o: dict[str, Any]) -> str: if o["paid_at"] is not None: stamp = o["paid_at"].strftime("%Y-%m-%d %H:%M:%S") @@ -212,7 +229,7 @@ async def _setup(t: Target) -> None: if t.mysql: # 关掉严格模式才写得进零日期（老库里常见的情形） await conn.execute(text("SET SESSION sql_mode = ''")) - for sql in _ddl(t) + _rows(t) + _events(1, N_EVENTS_OLD): + for sql in _ddl(t) + _rows(t) + _events(1, N_EVENTS_OLD) + _snapshot(t): await conn.exec_driver_sql(sql) async with engine.begin() as conn: for sql in _analyze(t, "events") + _analyze(t, "orders") + _analyze(t, "stores"): @@ -421,10 +438,41 @@ async def run_target(t: Target) -> None: res = await data_engine.run_query(source, d.max_date_sql(tbl, paid_col)) check(bool(res.rows) and str(res.rows[0][0])[:19].replace("T", " ") == str(EXPECT["paid_at"][1]), f"max_date_sql 直接跑：{res.rows}") + await _coverage(t, source) finally: await data_engine.engines.invalidate(source.id) +async def _coverage(t: Target, source: Any) -> None: + """覆盖检查（data/coverage.py）在真库上：按运行当天现算的上周区间，表里只有当天的数据 → 判成没有数据；区间包含当天、 + 只是别的条件筛空的 → 不报。界在库里算（各方言自己的写法），表的范围现查 MIN / MAX。""" + from app.data import coverage + + last_week = {"mysql": ("event_day >= DATE_SUB(CURRENT_DATE, INTERVAL (WEEKDAY(CURRENT_DATE) + 7) DAY) " + "AND event_day < DATE_SUB(CURRENT_DATE, INTERVAL WEEKDAY(CURRENT_DATE) DAY)"), + "pg": ("event_day >= date_trunc('week', CURRENT_DATE) - INTERVAL '7 days' " + "AND event_day < date_trunc('week', CURRENT_DATE)"), + "oracle": "event_day >= TRUNC(SYSDATE, 'IW') - 7 AND event_day < TRUNC(SYSDATE, 'IW')"} + recent = {"mysql": "event_day >= DATE_SUB(CURRENT_DATE, INTERVAL 1 DAY)", + "pg": "event_day >= CURRENT_DATE - INTERVAL '1 day'", + "oracle": "event_day >= TRUNC(SYSDATE) - 1"} + flavor = "mysql" if t.mysql else "pg" if t.pg else "oracle" + today_sql = "SELECT TRUNC(SYSDATE) FROM DUAL" if t.oracle else "SELECT CURRENT_DATE" + today = str((await data_engine.run_query(source, today_sql)).rows[0][0])[:10] + for table in ["daily_snapshot_d"] + (["daily_snapshot"] if t.mysql else []): + sql = f"SELECT COUNT(DISTINCT visitor) AS n FROM {table} WHERE {last_week[flavor]}" + gaps = await coverage.result_gaps(source, sql, (await data_engine.run_query(source, sql)).to_payload()) + check(len(gaps) == 1 and gaps[0]["data_range"] == [today, today] + and [c.lower() for c in gaps[0]["columns"]] == ["n"], + f"覆盖检查：{table} 只有当天的数据，上周的区间判成没有数据：{gaps}") + pre = await coverage.sql_gaps(source, sql) + check(len(pre) == 1, f"助手自查：{table} 上周的区间判成没有数据：{pre}") + fine = f"SELECT COUNT(DISTINCT visitor) AS n FROM {table} WHERE visitor = 'nobody' AND {recent[flavor]}" + quiet = await coverage.result_gaps(source, fine, (await data_engine.run_query(source, fine)).to_payload()) + check(quiet == [], f"覆盖检查：{table} 区间里有数据、只是别的条件筛空的不报：{quiet}") + check(await coverage.sql_gaps(source, fine) == [], f"助手自查：{table} 区间包含当天的不报") + + async def main() -> int: only = set((os.environ.get("DIALECT_IT_ONLY") or "pg,mysql,mariadb,oracle").split(",")) plan = [("pg", "postgresql", "DIALECT_IT_PG_PORT"), ("mysql", "mysql", "DIALECT_IT_MYSQL_PORT"),