yilunjiang/AgentLab
1
0
Fork 0
mirror of https://github.com/ThearchyHelios/AgentLab.git synced 2026-11-15 00:21:58 +00:00
Code Issues Packages Projects Releases Wiki Activity

覆盖检查：现查没成功（大表全表 MIN / MAX 超时、库一时连不上）也记 LIVE_TTL_S，期间同一张表直接退回剖析、取值观察，不再每次白等 PROBE_TIMEOUT_S 秒（测试 test_a_failed_probe_is_not_retried_for_a_while）；方言容器集成测试（scripts/dialect_it.py）加一段覆盖检查：只存当天的滚动表（DATE 列，MySQL / MariaDB 另有按文字 'YYYYMMDD' 存的），按运行当天现算的上周区间判成没有数据、区间包含当天只是别的条件筛空的不报，查询后和助手自查两个入口都核——PostgreSQL 16、MySQL 8.4、MariaDB 11、Oracle Free 23 全过

Browse Source
This commit is contained in:
ThearchyHelios
2026-10-20 10:58:20 +08:00
parent b0556e36d6
commit 680ea4ecf8
3 changed files with 97 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/data/coverage.py
+19 -7
View File
@@ -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:
backend/tests/test_query_coverage.py
+29
View File
@@ -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, "过期了就重新现查"
scripts/dialect_it.py
+49 -1
View File
@@ -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"),