This is an automated email from the ASF dual-hosted git repository.
FreeOnePlus pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris-mcp-server.git
The following commit(s) were added to refs/heads/master by this push:
new fae518e fix: handle unknown data freshness safely (#98)
fae518e is described below
commit fae518e247d177a49f2d3c2607f4c1d386f88222
Author: Yijia Su <[email protected]>
AuthorDate: Wed Jul 29 17:47:43 2026 +0800
fix: handle unknown data freshness safely (#98)
Co-authored-by: FreeOnePlus <[email protected]>
---
MCP-2026-07-28-DEVELOPMENT-LEDGER.md | 37 ++++++++++++-
doris_mcp_server/tools/tools_manager.py | 2 +
doris_mcp_server/utils/data_governance_tools.py | 18 ++++---
test/protocol/stdio_capability_server.py | 43 ++++++++++++++-
test/protocol/test_mcp_v2_protocol.py | 70 +++++++++++++++++++++++++
test/tools/test_tools_operation_guard.py | 40 ++++++++++++++
6 files changed, 201 insertions(+), 9 deletions(-)
diff --git a/MCP-2026-07-28-DEVELOPMENT-LEDGER.md
b/MCP-2026-07-28-DEVELOPMENT-LEDGER.md
index 005bcb3..19229d3 100644
--- a/MCP-2026-07-28-DEVELOPMENT-LEDGER.md
+++ b/MCP-2026-07-28-DEVELOPMENT-LEDGER.md
@@ -139,7 +139,7 @@
| `CORE-008` | P2 | 大结果边界 | 无 | 行数、字节数、超时和取消可配置且有硬上限 | `BACKLOG` |
| `CORE-009` | P2 | manager 模块职责拆分 | CORE-005 | 不改变行为前提下缩小超大文件，模块边界有测试 |
`BACKLOG` |
| `CORE-010` | P1 | 修复 SQL profile 分析未绑定 `auth_context` | 无 | 真实 Doris
调用不再触发局部变量未赋值；鉴权上下文覆盖测试通过 | `DONE` |
-| `CORE-011` | P1 | 修复数据新鲜度空阈值比较 | 无 | 阈值缺失或为 `None` 时返回类型化错误/默认值，不抛
`TypeError` | `READY` |
+| `CORE-011` | P1 | 修复数据新鲜度空阈值比较 | 无 | 阈值缺失或为 `None` 时返回类型化错误/默认值，不抛
`TypeError` | `DONE` |
| `COMPAT-001` | P1 | Doris 4.0 元数据字段兼容 | 无 | Doris 4.0.5 的角色/权限查询不再依赖不存在的
`Default_role` 字段 | `READY` |
| `COMPAT-002` | P2 | FE/BE HTTP 端点独立配置 | SEC-018 | SQL、FE HTTP、BE HTTP
可分别配置主机/端口并通过代理/隧道环境测试 | `BACKLOG` |
@@ -477,13 +477,46 @@ recovery result: org_tenant=47040
连接通过既有 SSH key 与临时 SQL/FE HTTP 隧道完成；凭据未写入仓库、测试或台账，探针结束后 MCP 服务与隧道均已关闭。
+提交与评审回执：
+
+- commit：`e2e90e0 fix: bind auth context for SQL profiles`
+- Draft
PR：[apache/doris-mcp-server#97](https://github.com/apache/doris-mcp-server/pull/97)
+
+### CORE-011
+
+数据新鲜度工具现在把缺省或显式为 `None` 的 `freshness_threshold_hours` 统一归一为 24
小时，并在业务实现入口再次兜底。无法从 Doris 推断更新时间时，`staleness_hours=None` 保持为明确的 `unknown`
状态，不再进入数值比较；只有真实数字且超过 72 小时时才生成“严重陈旧”数据流问题。
+
+自动化验证：
+
+- 路由层：显式 `None` 向业务层传递默认值 24；
+- 分析层：`unknown + staleness_hours=None` 返回空问题列表，不抛 `TypeError`；
+- Streamable HTTP：缺省阈值、未知更新时间返回结构化结果；
+- 真实子进程 STDIO modern/legacy：同一未知状态和默认阈值契约；
+- 完整 pytest：`351 passed / 57 skipped / 0 failed / 249 warnings`；
+- `uv lock --check`、新增测试完整 Ruff、运行时严重错误规则 Ruff、`compileall`、`uv build` 全部通过。
+
+真实 Doris 验证：
+
+```text
+environment: 192.168.31.63 / hhm_dt_sim
+table: org_tenant
+default threshold: 24
+explicit null threshold: 24
+freshness status: unknown
+recovery result: org_tenant=47040
+```
+
+- HTTP modern/legacy：缺省与显式空阈值均返回 24，未知更新时间不抛异常，随后真实查询成功；
+- STDIO modern/legacy：同样覆盖缺省与显式空阈值，随后真实查询成功；
+- 连接通过既有 SSH key 和临时本地隧道完成；凭据未写入仓库、测试或台账，探针完成后服务与隧道均已关闭。
+
## 11. 下一开发批次
批次：`BATCH-02-CONFORMANCE-AND-ERROR-SEMANTICS`
按以下顺序推进：
-1. `CORE-011` / `COMPAT-001`：继续修复真实 Doris 已复现缺陷；
+1. `COMPAT-001`：继续修复真实 Doris 已复现缺陷；
2. `TEST-003`：运行官方 `server-stateless` Conformance；
3. `TEST-005` / `TEST-012`：补权限不足、超时、故障恢复和工具错误路径；
4. `PROTO-018` / `DOC-001` / `DOC-002`：版本单一来源和迁移文档；
diff --git a/doris_mcp_server/tools/tools_manager.py
b/doris_mcp_server/tools/tools_manager.py
index 8d12c31..89e5e30 100644
--- a/doris_mcp_server/tools/tools_manager.py
+++ b/doris_mcp_server/tools/tools_manager.py
@@ -1844,6 +1844,8 @@ No parameters required. Returns connection status,
configuration, and diagnostic
"""Data freshness monitoring tool routing"""
table_names = arguments.get("table_names")
freshness_threshold_hours = arguments.get("freshness_threshold_hours",
24)
+ if freshness_threshold_hours is None:
+ freshness_threshold_hours = 24
include_update_patterns = arguments.get("include_update_patterns",
True)
catalog_name = arguments.get("catalog_name")
db_name = arguments.get("db_name")
diff --git a/doris_mcp_server/utils/data_governance_tools.py
b/doris_mcp_server/utils/data_governance_tools.py
index ebdc11c..4e3238d 100644
--- a/doris_mcp_server/utils/data_governance_tools.py
+++ b/doris_mcp_server/utils/data_governance_tools.py
@@ -162,6 +162,8 @@ class DataGovernanceTools:
"""
try:
start_time = time.time()
+ if time_threshold_hours is None:
+ time_threshold_hours = 24
connection = await self.connection_manager.get_connection("query")
# 1. Get list of tables to monitor
@@ -967,11 +969,15 @@ class DataGovernanceTools:
})
# Identify particularly stale tables
- very_stale_tables = [
- (table_name, info.get("staleness_hours", 0))
- for table_name, info in table_freshness.items()
- if info.get("staleness_hours", 0) > 72 # More than 3 days
- ]
+ very_stale_tables = []
+ for table_name, info in table_freshness.items():
+ staleness_hours = info.get("staleness_hours")
+ if (
+ isinstance(staleness_hours, (int, float))
+ and not isinstance(staleness_hours, bool)
+ and staleness_hours > 72
+ ):
+ very_stale_tables.append((table_name, staleness_hours))
if very_stale_tables:
issues.append({
@@ -1019,4 +1025,4 @@ class DataGovernanceTools:
"timestamp": datetime.now().isoformat()
})
- return alerts
\ No newline at end of file
+ return alerts
diff --git a/test/protocol/stdio_capability_server.py
b/test/protocol/stdio_capability_server.py
index ce15db8..0991a46 100644
--- a/test/protocol/stdio_capability_server.py
+++ b/test/protocol/stdio_capability_server.py
@@ -34,7 +34,9 @@ from mcp.types import (
)
from doris_mcp_server.protocol import create_doris_mcp_server
+from doris_mcp_server.tools.tools_manager import DorisToolsManager
from doris_mcp_server.utils.analysis_tools import SQLAnalyzer
+from doris_mcp_server.utils.data_governance_tools import DataGovernanceTools
from doris_mcp_server.utils.db import QueryResult
REQUIRED_EXTENSION = "io.apache.doris/read"
@@ -88,9 +90,30 @@ class ProfileAnalyzer(SQLAnalyzer):
return {"profile": "ok", "query_id": query_id}
+class FreshnessGovernanceTools(DataGovernanceTools):
+ async def _analyze_table_freshness(
+ self,
+ connection,
+ table_name: str,
+ threshold_hours: int,
+ ) -> dict:
+ return {
+ "last_update": None,
+ "staleness_hours": None,
+ "freshness_score": 0.0,
+ "status": "unknown",
+ "method_used": "none",
+ }
+
+
class OneToolManager:
def __init__(self) -> None:
- self.profile_analyzer = ProfileAnalyzer(ProfileConnectionManager())
+ connection_manager = ProfileConnectionManager()
+ self.profile_analyzer = ProfileAnalyzer(connection_manager)
+ self.freshness_router = object.__new__(DorisToolsManager)
+ self.freshness_router.data_governance_tools = FreshnessGovernanceTools(
+ connection_manager
+ )
async def list_tools(self) -> list[Tool]:
return [
@@ -111,6 +134,20 @@ class OneToolManager:
"required": ["sql"],
},
),
+ Tool(
+ name="monitor_data_freshness",
+ description="Exercise unknown freshness values with the
default threshold.",
+ input_schema={
+ "type": "object",
+ "properties": {
+ "table_names": {
+ "type": "array",
+ "items": {"type": "string"},
+ },
+ "freshness_threshold_hours": {"type": "integer"},
+ },
+ },
+ ),
]
async def call_tool(self, name: str, arguments: dict) -> str:
@@ -121,6 +158,10 @@ class OneToolManager:
db_name=arguments.get("db_name"),
)
)
+ if name == "monitor_data_freshness":
+ return json.dumps(
+ await
self.freshness_router._monitor_data_freshness_tool(arguments)
+ )
return "{}"
diff --git a/test/protocol/test_mcp_v2_protocol.py
b/test/protocol/test_mcp_v2_protocol.py
index 7741877..624a419 100644
--- a/test/protocol/test_mcp_v2_protocol.py
+++ b/test/protocol/test_mcp_v2_protocol.py
@@ -664,6 +664,43 @@ async def
test_http_sql_profile_without_catalog_uses_production_analyzer_path():
assert recovered.json()["result"]["isError"] is False
[email protected]
+async def test_http_unknown_freshness_uses_default_threshold():
+ app = create_test_server(
+ tools_manager=ProfileToolManager(),
+ ).streamable_http_app(
+ json_response=True,
+ stateless_http=True,
+ host="127.0.0.1",
+ transport_security=create_transport_security("127.0.0.1"),
+ )
+
+ async with (
+ app.router.lifespan_context(app),
+ httpx2.ASGITransport(app) as transport,
+ httpx2.AsyncClient(
+ transport=transport,
+ base_url="http://127.0.0.1:3000";,
+ ) as client,
+ ):
+ result = await client.post(
+ "/mcp",
+ json=modern_tool_request(
+ 1,
+ "monitor_data_freshness",
+ {"table_names": ["org_tenant"]},
+ ),
+ headers=modern_tool_headers("monitor_data_freshness"),
+ )
+
+ assert result.status_code == 200
+ payload = result.json()["result"]["structuredContent"]
+ assert result.json()["result"]["isError"] is False
+ assert "error" not in payload
+ assert payload["monitoring_scope"]["time_threshold_hours"] == 24
+ assert payload["table_freshness"]["org_tenant"]["status"] == "unknown"
+
+
@pytest.mark.asyncio
async def test_stdio_validates_capabilities_versions_and_process_survival():
server_script = Path(__file__).with_name("stdio_capability_server.py")
@@ -713,12 +750,14 @@ async def
test_stdio_validates_capabilities_versions_and_process_survival():
assert [tool.name for tool in (await capable.list_tools()).tools] == [
"echo",
"get_sql_profile",
+ "monitor_data_freshness",
]
async with Client(stdio_client(server_params), mode="legacy") as legacy:
assert [tool.name for tool in (await legacy.list_tools()).tools] == [
"echo",
"get_sql_profile",
+ "monitor_data_freshness",
]
legacy_error = await legacy.read_resource("doris://table/missing")
assert json.loads(legacy_error.contents[0].text)["error_code"] ==
"RESOURCE_NOT_FOUND"
@@ -755,6 +794,37 @@ async def
test_stdio_sql_profile_without_catalog_uses_production_analyzer_path()
assert json.loads(result.content[0].text)["success"] is True
[email protected]
+async def test_stdio_unknown_freshness_uses_default_threshold():
+ server_script = Path(__file__).with_name("stdio_capability_server.py")
+ server_params = StdioServerParameters(
+ command=sys.executable,
+ args=[str(server_script)],
+ )
+
+ async with Client(
+ stdio_client(server_params),
+ extensions=[advertise(REQUIRED_EXTENSION)],
+ ) as modern:
+ result = await modern.call_tool(
+ "monitor_data_freshness",
+ {"table_names": ["org_tenant"]},
+ )
+ payload = result.structured_content
+ assert result.is_error is False
+ assert "error" not in payload
+ assert payload["monitoring_scope"]["time_threshold_hours"] == 24
+
+ async with Client(stdio_client(server_params), mode="legacy") as legacy:
+ result = await legacy.call_tool(
+ "monitor_data_freshness",
+ {"table_names": ["org_tenant"]},
+ )
+ payload = json.loads(result.content[0].text)
+ assert "error" not in payload
+ assert payload["monitoring_scope"]["time_threshold_hours"] == 24
+
+
@pytest.mark.asyncio
async def test_stdio_prompt_errors_are_typed_and_process_survives():
server_script = Path(__file__).with_name("stdio_capability_server.py")
diff --git a/test/tools/test_tools_operation_guard.py
b/test/tools/test_tools_operation_guard.py
index 8876a6f..b7dd71d 100644
--- a/test/tools/test_tools_operation_guard.py
+++ b/test/tools/test_tools_operation_guard.py
@@ -12,6 +12,7 @@ from doris_mcp_server.auth.operation_policy import (
from doris_mcp_server.protocol import create_doris_mcp_server
from doris_mcp_server.tools.tools_manager import DorisToolsManager
from doris_mcp_server.utils.analysis_tools import SQLAnalyzer
+from doris_mcp_server.utils.data_governance_tools import DataGovernanceTools
from doris_mcp_server.utils.db import QueryResult
from doris_mcp_server.utils.query_executor import DorisQueryExecutor
from doris_mcp_server.utils.schema_extractor import MetadataExtractor
@@ -410,6 +411,45 @@ async def
test_sql_profile_binds_auth_context_without_catalog(tmp_path, db_name)
assert call["doris_user"] == "alice"
[email protected]
+async def test_data_freshness_none_threshold_uses_default():
+ manager = object.__new__(DorisToolsManager)
+ manager.data_governance_tools = SimpleNamespace(
+ monitor_data_freshness=AsyncMock(return_value={"ok": True})
+ )
+
+ result = await manager._monitor_data_freshness_tool(
+ {
+ "table_names": ["orders"],
+ "freshness_threshold_hours": None,
+ }
+ )
+
+ assert result == {"ok": True}
+
manager.data_governance_tools.monitor_data_freshness.assert_awaited_once_with(
+ tables=["orders"],
+ time_threshold_hours=24,
+ catalog_name=None,
+ db_name=None,
+ )
+
+
[email protected]
+async def test_unknown_data_freshness_is_not_compared_as_a_number():
+ governance = DataGovernanceTools(SimpleNamespace())
+
+ issues = await governance._identify_data_flow_issues(
+ {
+ "orders": {
+ "status": "unknown",
+ "staleness_hours": None,
+ }
+ }
+ )
+
+ assert issues == []
+
+
@pytest.mark.asyncio
async def
test_doris_oauth_list_tools_uses_configured_default_scope_visibility(tmp_path):
manager, _connection_manager = _real_tool_manager_for_routing(tmp_path)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
