"""EXPLAIN 执行服务 — 连接数据源执行 EXPLAIN 并解析为统一树结构。 支持: - MySQL: EXPLAIN FORMAT=JSON - PostgreSQL: EXPLAIN (FORMAT JSON, ANALYZE false) - SQLite: EXPLAIN QUERY PLAN 返回统一树结构: { "execution_plan": [...], # 树形执行计划节点 "warnings": [...], # 优化警告 "summary": {...}, # 汇总统计 "db_type": "mysql", "raw_result": {...}, # 原始结果 } """ import asyncio import json import logging from typing import Any, Dict, List, Optional, Tuple # 顶层 import 以便测试 mock（实际运行时为可选依赖） try: import aiomysql # noqa: F401 except ImportError: aiomysql = None # type: ignore[assignment] try: import asyncpg # noqa: F401 except ImportError: asyncpg = None # type: ignore[assignment] from app.models.instance import Instance logger = logging.getLogger(__name__) # ===== 统一节点结构 ===== def make_plan_node( id: int, operation: str, table: Optional[str] = None, rows: Optional[int] = None, cost: Optional[float] = None, key: Optional[str] = None, extra: Optional[str] = None, children: Optional[List[Dict]] = None, **kwargs, ) -> Dict[str, Any]: """构建统一执行计划节点。""" node = { "id": id, "operation": operation, "table": table, "rows": rows, "cost": cost, "key": key, "extra": extra, "children": children or [], } # 附加额外字段 for k, v in kwargs.items(): if v is not None: node[k] = v return node # ===== MySQL EXPLAIN FORMAT=JSON 解析 ===== def parse_mysql_explain_json(raw_json: Any) -> Dict[str, Any]: """解析 MySQL EXPLAIN FORMAT=JSON 的结果。 MySQL EXPLAIN FORMAT=JSON 结构示例: { "query_block": { "select_id": 1, "cost_info": {"query_cost": "0.35"}, "table": { "table_name": "users", "access_type": "ALL", "rows_examined_per_scan": 1000, "rows_produced_per_join": 1000, "filtered": "100.00", "cost_info": {...}, "used_columns": [...] }, "nested_loop": [...] } } Returns: 统一结构的 Dict """ if isinstance(raw_json, str): try: raw_json = json.loads(raw_json) except (json.JSONDecodeError, TypeError): return _empty_result("mysql", raw_json, error="EXPLAIN JSON 解析失败") nodes: List[Dict] = [] warnings: List[str] = [] node_id_counter = [0] def next_id() -> int: node_id_counter[0] += 1 return node_id_counter[0] def parse_query_block(qb: Dict) -> Dict[str, Any]: """递归解析 query_block。""" node_id = next_id() children = [] operation = "Query Block" table_name = None access_type = None key_used = None rows_est = None cost_val = None extra_parts = [] # 表信息 table_info = qb.get("table") if table_info: table_name = table_info.get("table_name") access_type = table_info.get("access_type", "") key_used = table_info.get("key") possible_keys = table_info.get("possible_keys") rows_est = table_info.get("rows_examined_per_scan") filtered = table_info.get("filtered", "100.00") # 成本 cost_info = table_info.get("cost_info", {}) cost_str = cost_info.get("read_cost") or cost_info.get("eval_cost") or cost_info.get("prefix_cost") if cost_str: try: cost_val = float(cost_str) except (ValueError, TypeError): pass # 操作类型映射 operation = _mysql_access_type_to_operation(access_type) # Extra 信息 if key_used: extra_parts.append(f"使用索引: {key_used}") if possible_keys and not key_used: extra_parts.append(f"可选索引: {possible_keys}") if filtered and filtered != "100.00": extra_parts.append(f"过滤率: {filtered}%") used_key_parts = table_info.get("used_key_parts") if used_key_parts: extra_parts.append(f"索引列: {', '.join(str(p) for p in used_key_parts)}") # 警告检测 if access_type == "ALL": warnings.append(f"表 '{table_name}' 全表扫描，建议添加索引") if access_type == "index": warnings.append(f"表 '{table_name}' 全索引扫描") # 嵌套循环 / join nested_loop = qb.get("nested_loop") if nested_loop: for nl_item in nested_loop: child = parse_query_block(nl_item) children.append(child) # 排序 order_by = qb.get("ordering_operation") if order_by: sort_node = _parse_mysql_ordering(order_by, next_id()) children.append(sort_node) extra_parts.append("Using filesort") warnings.append("使用了文件排序 (filesort)") # 分组 grouping = qb.get("grouping_operation") if grouping: group_node = _parse_mysql_grouping(grouping, next_id()) children.append(group_node) extra_parts.append("Using temporary") warnings.append("使用了临时表 (temporary)") # 子查询 subqueries = qb.get("subqueries") or qb.get("optimized_away_subqueries") if subqueries: for sq in subqueries: if "query_block" in sq: child = parse_query_block(sq["query_block"]) children.append(child) # 成本信息 qb_cost = qb.get("cost_info", {}) if not cost_val and "query_cost" in qb_cost: try: cost_val = float(qb_cost["query_cost"]) except (ValueError, TypeError): pass return make_plan_node( id=node_id, operation=operation, table=table_name, rows=int(rows_est) if rows_est else None, cost=cost_val, key=key_used, extra="; ".join(extra_parts) if extra_parts else None, children=children, access_type=access_type, ) # 解析顶层 query_block query_block = raw_json.get("query_block") if isinstance(raw_json, dict) else None if query_block: root = parse_query_block(query_block) nodes = [root] else: warnings.append("无法识别的 EXPLAIN JSON 结构") # 汇总 summary = _build_summary(nodes, warnings, "mysql") return { "execution_plan": nodes, "warnings": warnings, "summary": summary, "db_type": "mysql", "raw_result": raw_json, } def _parse_mysql_ordering(ordering: Dict, node_id: int) -> Dict: """解析 MySQL 排序操作。""" table_info = ordering.get("table", {}) return make_plan_node( id=node_id, operation="Filesort", table=table_info.get("table_name"), extra="排序操作", ) def _parse_mysql_grouping(grouping: Dict, node_id: int) -> Dict: """解析 MySQL 分组操作。""" table_info = grouping.get("table", {}) return make_plan_node( id=node_id, operation="Temporary Table", table=table_info.get("table_name"), extra="分组操作使用临时表", ) def _mysql_access_type_to_operation(access_type: Optional[str]) -> str: """将 MySQL access_type 映射为统一操作名。""" mapping = { "system": "System Scan", "const": "Const Row", "eq_ref": "Unique Lookup", "ref": "Index Lookup", "fulltext": "Fulltext Search", "ref_or_null": "Index Lookup (nullable)", "index_merge": "Index Merge", "unique_subquery": "Unique Subquery", "index_subquery": "Index Subquery", "range": "Index Range Scan", "index": "Full Index Scan", "ALL": "Table Scan", } return mapping.get(access_type or "", f"Unknown ({access_type})") # ===== PostgreSQL EXPLAIN (FORMAT JSON) 解析 ===== def parse_postgres_explain_json(raw_json: Any) -> Dict[str, Any]: """解析 PostgreSQL EXPLAIN (FORMAT JSON) 的结果。 PostgreSQL EXPLAIN JSON 结构示例: [ { "Plan": { "Node Type": "Seq Scan", "Relation Name": "users", "Alias": "users", "Startup Cost": 0.00, "Total Cost": 15.50, "Plan Rows": 1000, "Plan Width": 68, "Plans": [ { "Node Type": "Index Scan", ... "Plans": [] } ] } } ] Returns: 统一结构的 Dict """ if isinstance(raw_json, str): try: raw_json = json.loads(raw_json) except (json.JSONDecodeError, TypeError): return _empty_result("postgres", raw_json, error="EXPLAIN JSON 解析失败") # PG 返回的是数组 if isinstance(raw_json, list) and len(raw_json) > 0: plan_root = raw_json[0].get("Plan", raw_json[0]) elif isinstance(raw_json, dict): plan_root = raw_json.get("Plan", raw_json) else: return _empty_result("postgres", raw_json, error="无法识别的 EXPLAIN JSON 结构") warnings: List[str] = [] node_id_counter = [0] def next_id() -> int: node_id_counter[0] += 1 return node_id_counter[0] def parse_pg_plan(plan: Dict) -> Dict[str, Any]: """递归解析 PostgreSQL Plan 节点。""" node_id = next_id() children = [] node_type = plan.get("Node Type", "Unknown") relation_name = plan.get("Relation Name") or plan.get("Table Name") alias = plan.get("Alias") startup_cost = plan.get("Startup Cost") total_cost = plan.get("Total Cost") plan_rows = plan.get("Plan Rows") plan_width = plan.get("Plan Width") index_name = plan.get("Index Name") index_cond = plan.get("Index Cond") filter_str = plan.get("Filter") hash_cond = plan.get("Hash Cond") join_type = plan.get("Join Type") sort_key = plan.get("Sort Key") sort_method = plan.get("Sort Method") group_key = plan.get("Group Key") actual_rows = plan.get("Actual Rows") actual_time = plan.get("Actual Total Time") # 操作类型映射 operation = _pg_node_type_to_operation(node_type) if join_type: operation = f"{join_type} {_pg_node_type_to_operation(node_type)}" # Extra 信息 extra_parts = [] if index_cond: extra_parts.append(f"索引条件: {index_cond}") if filter_str: extra_parts.append(f"过滤条件: {filter_str}") if hash_cond: extra_parts.append(f"Hash 条件: {hash_cond}") if sort_key: extra_parts.append(f"排序键: {sort_key}") if sort_method: extra_parts.append(f"排序方法: {sort_method}") if group_key: extra_parts.append(f"分组键: {group_key}") if actual_rows is not None: extra_parts.append(f"实际行数: {actual_rows}") if actual_time is not None: extra_parts.append(f"实际耗时: {actual_time}ms") # 警告检测 if node_type == "Seq Scan": warnings.append(f"表 '{relation_name or '?'}' 使用了顺序扫描 (Seq Scan)，建议检查索引") if node_type == "Sort" and sort_method == "external" or plan.get("Sort Space Used"): warnings.append(f"排序使用了外部排序 (磁盘)，建议增加 work_mem") # 子计划 sub_plans = plan.get("Plans", []) for sp in sub_plans: children.append(parse_pg_plan(sp)) # 子查询/SubPlan subplan_name = plan.get("Subplan Name") if subplan_name: extra_parts.append(f"子计划: {subplan_name}") kwargs = {} if alias and alias != relation_name: kwargs["alias"] = alias if plan_width: kwargs["plan_width"] = plan_width if startup_cost is not None: kwargs["startup_cost"] = startup_cost if actual_time is not None: kwargs["actual_time_ms"] = actual_time if actual_rows is not None: kwargs["actual_rows"] = actual_rows return make_plan_node( id=node_id, operation=operation, table=relation_name, rows=int(plan_rows) if plan_rows else None, cost=float(total_cost) if total_cost is not None else None, key=index_name, extra="; ".join(extra_parts) if extra_parts else None, children=children, node_type=node_type, **kwargs, ) root = parse_pg_plan(plan_root) nodes = [root] # 汇总 summary = _build_summary(nodes, warnings, "postgres") return { "execution_plan": nodes, "warnings": warnings, "summary": summary, "db_type": "postgres", "raw_result": raw_json, } def _pg_node_type_to_operation(node_type: str) -> str: """将 PostgreSQL Node Type 映射为统一操作名。""" mapping = { "Seq Scan": "Table Scan", "Index Scan": "Index Scan", "Index Only Scan": "Index Only Scan", "Bitmap Heap Scan": "Bitmap Heap Scan", "Bitmap Index Scan": "Bitmap Index Scan", "Tid Scan": "TID Scan", "Subquery Scan": "Subquery Scan", "Function Scan": "Function Scan", "Table Function Scan": "Table Function Scan", "Values Scan": "Values Scan", "CTE Scan": "CTE Scan", "Named Tuplestore Scan": "Named Tuplestore Scan", "WorkTable Scan": "WorkTable Scan", "Nested Loop": "Nested Loop Join", "Merge Join": "Merge Join", "Hash Join": "Hash Join", "Sort": "Sort", "Hash": "Hash", "Aggregate": "Aggregate", "Group": "Group", "Append": "Append", "Merge Append": "Merge Append", "Limit": "Limit", "Unique": "Unique", "Gather": "Gather", "Gather Merge": "Gather Merge", "Materialize": "Materialize", "Result": "Result", "ProjectSet": "ProjectSet", "WindowAgg": "Window Aggregate", "Recursive Union": "Recursive Union", "LockRows": "Lock Rows", "ModifyTable": "Modify Table", "SetOp": "Set Operation", } return mapping.get(node_type, node_type) # ===== SQLite EXPLAIN QUERY PLAN 解析 ===== def parse_sqlite_explain(rows: List[Dict[str, Any]]) -> Dict[str, Any]: """解析 SQLite EXPLAIN QUERY PLAN 的文本/行结果。 SQLite 返回格式: id|parent|notused|detail 3|0|0|SCAN TABLE users 或 id|parent|notused|detail 2|0|0|SEARCH TABLE users USING INDEX idx_name (name=?) Returns: 统一结构的 Dict """ warnings: List[str] = [] node_id_counter = [0] node_map: Dict[int, Dict] = {} def next_id() -> int: node_id_counter[0] += 1 return node_id_counter[0] for row in rows: # SQLite EXPLAIN QUERY PLAN 返回: id, parent, notused, detail detail = row.get("detail") or row.get("DETAIL") or "" parent_id = row.get("parent") or row.get("PARENT") or 0 sqlite_id = row.get("id") or row.get("ID") or next_id() table_name = None operation = "Unknown" key = None extra = None detail_upper = detail.upper() # 解析 detail 文本 if "SCAN TABLE" in detail_upper: operation = "Table Scan" # 提取表名: "SCAN TABLE users" parts = detail.split("TABLE", 1) if len(parts) > 1: table_part = parts[1].strip().split()[0] if parts[1].strip() else None table_name = table_part warnings.append(f"表 '{table_name or '?'}' 全表扫描，建议添加索引") elif "SEARCH TABLE" in detail_upper: operation = "Index Lookup" parts = detail.split("TABLE", 1) if len(parts) > 1: rest = parts[1].strip() tokens = rest.split() table_name = tokens[0] if tokens else None # 提取索引名: "USING INDEX idx_name (col=?)" if "USING INDEX" in detail_upper: idx_parts = detail.split("USING INDEX", 1) if len(idx_parts) > 1: idx_rest = idx_parts[1].strip().split("(")[0].strip() key = idx_rest elif "USING COVERING INDEX" in detail_upper: idx_parts = detail.split("USING COVERING INDEX", 1) if len(idx_parts) > 1: idx_rest = idx_parts[1].strip().split("(")[0].strip() key = idx_rest extra = "Covering Index" elif "EXECUTE" in detail_upper: operation = "Execute" extra = detail elif "CORRELATED SCALAR SUBQUERY" in detail_upper: operation = "Correlated Subquery" extra = detail elif "CO-ROUTINE" in detail_upper: operation = "Co-routine" extra = detail elif "SCAN" in detail_upper and "TABLE" not in detail_upper: operation = "Scan" extra = detail else: operation = detail if detail else "Unknown" node = make_plan_node( id=next_id(), operation=operation, table=table_name, key=key, extra=extra, ) node_map[sqlite_id] = node # 构建树（基于 parent） root_nodes = [] for sqlite_id, node in node_map.items(): parent = None for row in rows: rid = row.get("id") or row.get("ID") if rid == sqlite_id: parent = row.get("parent") or row.get("PARENT") or 0 break if parent == 0 or parent not in node_map: root_nodes.append(node) else: parent_node = node_map[parent] parent_node["children"].append(node) if not root_nodes and node_map: root_nodes = list(node_map.values()) summary = _build_summary(root_nodes, warnings, "sqlite") return { "execution_plan": root_nodes, "warnings": warnings, "summary": summary, "db_type": "sqlite", "raw_result": rows, } # ===== 辅助函数 ===== def _empty_result(db_type: str, raw: Any, error: str = "") -> Dict[str, Any]: """返回空结果（解析失败时）。""" return { "execution_plan": [], "warnings": [error] if error else ["EXPLAIN 结果为空"], "summary": { "total_cost": 0, "estimated_rows": 0, "has_full_scan": False, "has_temporary": False, "has_filesort": False, "index_usage_ratio": 0.0, }, "db_type": db_type, "raw_result": raw, } def _build_summary( nodes: List[Dict], warnings: List[str], db_type: str, ) -> Dict[str, Any]: """从执行计划树构建汇总信息。""" total_cost = 0.0 total_rows = 0 has_full_scan = False has_temporary = False has_filesort = False index_used_count = 0 total_table_count = 0 def walk(node: Dict): nonlocal total_cost, total_rows, has_full_scan, has_temporary, has_filesort nonlocal index_used_count, total_table_count if node.get("cost") is not None: total_cost = max(total_cost, float(node["cost"])) if node.get("rows") is not None: total_rows += int(node["rows"]) op = (node.get("operation") or "").lower() extra = str(node.get("extra") or "").lower() if "table scan" in op or "seq scan" in op or op == "system scan": has_full_scan = True total_table_count += 1 elif "index" in op or "lookup" in op: index_used_count += 1 total_table_count += 1 elif "scan" in op: total_table_count += 1 if "temporary" in op or "temporary" in extra: has_temporary = True if "filesort" in op or "sort" in op or "filesort" in extra: has_filesort = True for child in node.get("children", []): walk(child) for n in nodes: walk(n) index_usage_ratio = 0.0 if total_table_count > 0: index_usage_ratio = round(index_used_count / total_table_count, 2) return { "total_cost": round(total_cost, 4), "estimated_rows": total_rows, "has_full_scan": has_full_scan, "has_temporary": has_temporary, "has_filesort": has_filesort, "index_usage_ratio": index_usage_ratio, } # ===== EXPLAIN 执行入口 ===== class ExplainService: """EXPLAIN 执行服务 — 统一入口。""" @staticmethod async def execute_explain( instance: Instance, database: Optional[str], query: str, ) -> Dict[str, Any]: """根据 db_type 构造并执行 EXPLAIN，返回统一树结构。 Args: instance: 数据源实例 database: 目标数据库名 query: 要解释的 SQL 查询 Returns: 统一 EXPLAIN 结果 Dict """ db_type = instance.db_type # 验证查询以 SELECT 或 WITH 开头 stripped = query.strip() upper = stripped.upper() if not upper.startswith(("SELECT", "WITH")): raise ValueError("EXPLAIN 仅支持 SELECT 或 WITH 查询") if db_type == "mysql": return await ExplainService._explain_mysql(instance, database, query) elif db_type in ("postgres", "postgresql"): return await ExplainService._explain_postgres(instance, database, query) elif db_type == "sqlite": return await ExplainService._explain_sqlite(instance, query) else: raise ValueError(f"暂不支持 {db_type} 类型的 EXPLAIN") @staticmethod async def _explain_mysql( instance: Instance, database: Optional[str], query: str ) -> Dict[str, Any]: """执行 MySQL EXPLAIN FORMAT=JSON。""" explain_sql = f"EXPLAIN FORMAT=JSON {query}" conn = await aiomysql.connect( host=instance.host, port=instance.port, user=instance.username, password=instance.password, db=database or instance.database, ) try: async with conn.cursor(aiomysql.DictCursor) as cur: await asyncio.wait_for(cur.execute(explain_sql), timeout=30) rows = await cur.fetchall() # MySQL EXPLAIN FORMAT=JSON 返回一行一列 JSON if rows: raw = rows[0].get("EXPLAIN") or rows[0].get("JSON") or rows[0] # raw 可能是字符串或已经解析的 dict if isinstance(raw, str): raw = json.loads(raw) else: raw = {} except asyncio.TimeoutError: raise ValueError("EXPLAIN 执行超时（30秒）") finally: conn.close() return parse_mysql_explain_json(raw) @staticmethod async def _explain_postgres( instance: Instance, database: Optional[str], query: str ) -> Dict[str, Any]: """执行 PostgreSQL EXPLAIN (FORMAT JSON)。""" explain_sql = f"EXPLAIN (FORMAT JSON, ANALYZE false) {query}" dsn = ( f"postgresql://{instance.username}:{instance.password}" f"@{instance.host}:{instance.port}/{database or instance.database or 'postgres'}" ) conn = await asyncpg.connect(dsn) try: result = await asyncio.wait_for(conn.fetchval(explain_sql), timeout=30) # asyncpg fetchval 返回 JSON 字符串 if isinstance(result, str): raw = json.loads(result) elif isinstance(result, list): raw = result else: raw = result except asyncio.TimeoutError: raise ValueError("EXPLAIN 执行超时（30秒）") finally: await conn.close() return parse_postgres_explain_json(raw) @staticmethod async def _explain_sqlite(instance: Instance, query: str) -> Dict[str, Any]: """执行 SQLite EXPLAIN QUERY PLAN。""" import sqlite3 explain_sql = f"EXPLAIN QUERY PLAN {query}" db_path = instance.host def _exec(): conn = sqlite3.connect(db_path) conn.row_factory = sqlite3.Row cur = conn.execute(explain_sql) rows = [dict(r) for r in cur.fetchall()] conn.close() return rows try: rows = await asyncio.wait_for( asyncio.get_event_loop().run_in_executor(None, _exec), timeout=30, ) except asyncio.TimeoutError: raise ValueError("EXPLAIN 执行超时（30秒）") return parse_sqlite_explain(rows)