覆盖检查现查到的表的范围在进程里记 5 分钟（按数据源、上传表格的快照、表、列认）：一张图里几个取数节点、助手自查几轮查同一张表，不再每次全表 MIN / MAX；查不成的、退回剖析的不记，过期了重新现查

backend/app/data/coverage.py

@@ -48,6 +48,10 @@ MAX_BOUNDS = 24
 MAX_SQL_CHARS = 20_000
 #: 一条 no_data 里最多列几个结果列（计数照实）
 MAX_COLUMNS = 12
+#: 现查到的表的范围在进程里记多久（秒）：同一张表一会儿内查好几次（一张图里几个取数节点、助手自查几轮）不再全表 MIN /
+#: MAX。滚动表一天才换一次数据，几分钟的旧不影响判断；查不成的、退回剖析的不记
+LIVE_TTL_S = 300.0
+_LIVE: dict[tuple[Any, ...], tuple[float, Any]] = {}
 
 #: 给人和模型看的说法
 NO_DATA_MESSAGE = "{columns}查的是 {window}，但 {table} 的 {column} {data}：这些 0 不代表这段时间真的是 0，是这段时间表里没有数据{basis}"
@@ -392,6 +396,11 @@ async def _range(source: Any, flt: _Filter, read: str | None, cache: Cache, fact
     key = (str(getattr(source, "name", "")), name_key(flt.table_name), name_key(flt.column_name))
     if key in cache.ranges:
         return cache.ranges[key]
+    live = (getattr(source, "id", None), getattr(source, "snapshot_id", None), *key)
+    held = _LIVE.get(live)
+    if held is not None and held[0] > time.monotonic():
+        cache.ranges[key] = held[1]
+        return held[1]
     table = flt.table.copy()
     table.set("alias", None)
     ident = flt.column.this.copy() if isinstance(flt.column.this, exp.Identifier) \
@@ -413,6 +422,10 @@ 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)
     return out
 
 

backend/tests/test_query_coverage.py

@@ -232,3 +232,28 @@ async def test_the_weekly_report_shape_on_mysql(monkeypatch):
     # 表的范围同一张表、同一列只查一次；按运行当天现算的界一条 SELECT 算出来
     assert sum("MIN(" in t for t in seen) == 1
     assert all(t.startswith("SELECT ") for t in seen)
+
+
+async def test_a_live_range_is_kept_for_a_while(monkeypatch):
+    """现查到的范围在进程里记几分钟：一张图里几个取数节点、助手自查几轮查同一张表，不再每次全表 MIN / MAX。"""
+    from types import SimpleNamespace
+
+    seen: list[str] = []
+
+    async def probe(source, text):
+        seen.append(text)
+        return ["20200101", "20200101"], ["text", "text"]
+
+    monkeypatch.setattr(coverage, "_probe", probe)
+    monkeypatch.setattr(coverage, "_LIVE", {})
+    source = SimpleNamespace(name=f"plant_{uuid.uuid4().hex[:6]}", kind="sqlite", schema_cache=None, id="s1")
+    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(2):
+        assert len(await coverage.result_gaps(source, sql, payload)) == 1
+    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)
+    await coverage.result_gaps(source, sql, payload)
+    assert sum("MIN(" in t for t in seen) == 3, "过期了就重新现查"