logan
f6f9142ce8
新增: - explain_preview.py: EXPLAIN预检服务(MySQL/PG), 影响行数预估, 全表扫描检测 - ticket.executor: 三种执行策略(stop_on_error/continue/transaction) - risk_evaluator: table_sizes参数, 大表ALTER/UPDATE/DELETE自动升级high - test_phase2_features.py: 22个测试(EXPLAIN+执行策略+大表检测) 简化: - 砍掉CTX-001上下文感知规则(复杂二次扫描), 合并大表检测到risk_evaluator - 审核引擎保持纯静态分析, 不再导入_RULE_REGISTRY做二次扫描 模型: - Ticket加exec_strategy字段 - TicketSqlItem模型 - schema加sql_type/exec_strategy/risk_level/workflow_id
336 lines
11 KiB
Python
336 lines
11 KiB
Python
"""EXPLAIN 预检服务 — 工单创建时跑 EXPLAIN 评估 SQL 影响。
|
||
|
||
支持:
|
||
- MySQL: EXPLAIN / EXPLAIN FORMAT=JSON
|
||
- PostgreSQL: EXPLAIN (FORMAT JSON, COSTS, BUFFERS)
|
||
- 影响行数预估
|
||
- 全表扫描检测
|
||
- 大表 ALTER 风险提示
|
||
"""
|
||
|
||
import json
|
||
import logging
|
||
import re
|
||
from dataclasses import dataclass, field
|
||
from typing import Any, Dict, List, Optional
|
||
|
||
from app.models.instance import Instance
|
||
from app.core.crypto import decrypt_password
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
@dataclass
|
||
class ExplainResult:
|
||
"""EXPLAIN 预检结果。"""
|
||
sql: str = ""
|
||
sql_type: str = "" # DQL/DML/DDL
|
||
estimated_rows: int = 0
|
||
full_table_scan: bool = False
|
||
using_index: bool = True
|
||
warnings: List[str] = field(default_factory=list)
|
||
explain_raw: Optional[str] = None
|
||
table_sizes: Dict[str, int] = field(default_factory=dict) # table_name -> row_count
|
||
error: Optional[str] = None
|
||
|
||
def to_dict(self) -> Dict[str, Any]:
|
||
return {
|
||
"sql": self.sql[:100],
|
||
"sql_type": self.sql_type,
|
||
"estimated_rows": self.estimated_rows,
|
||
"full_table_scan": self.full_table_scan,
|
||
"using_index": self.using_index,
|
||
"warnings": self.warnings,
|
||
"table_sizes": self.table_sizes,
|
||
"error": self.error,
|
||
}
|
||
|
||
|
||
# 大表阈值（行数）
|
||
LARGE_TABLE_THRESHOLD = 100_000
|
||
# 影响行数警告阈值
|
||
AFFECTED_ROWS_WARNING = 10_000
|
||
AFFECTED_ROWS_CRITICAL = 100_000
|
||
|
||
|
||
async def run_explain_preview(
|
||
sql: str,
|
||
instance: Instance,
|
||
target_database: str,
|
||
) -> ExplainResult:
|
||
"""对 SQL 执行 EXPLAIN 预检。
|
||
|
||
Args:
|
||
sql: SQL 语句
|
||
instance: 数据源实例
|
||
target_database: 目标数据库
|
||
|
||
Returns:
|
||
ExplainResult
|
||
"""
|
||
result = ExplainResult(sql=sql)
|
||
|
||
sql_upper = sql.strip().upper()
|
||
|
||
# 判断 SQL 类型
|
||
if sql_upper.startswith("SELECT"):
|
||
result.sql_type = "DQL"
|
||
elif sql_upper.startswith(("INSERT", "UPDATE", "DELETE")):
|
||
result.sql_type = "DML"
|
||
elif sql_upper.startswith(("CREATE", "ALTER", "DROP", "RENAME", "TRUNCATE")):
|
||
result.sql_type = "DDL"
|
||
else:
|
||
result.sql_type = "OTHER"
|
||
|
||
db_type = instance.db_type
|
||
|
||
try:
|
||
if db_type in ("mysql",):
|
||
result = await _explain_mysql(sql, instance, target_database, result)
|
||
elif db_type in ("postgres", "postgresql"):
|
||
result = await _explain_postgres(sql, instance, target_database, result)
|
||
else:
|
||
# Redis/MongoDB 不支持 EXPLAIN
|
||
result.warnings.append(f"{db_type} 不支持 EXPLAIN 预检")
|
||
return result
|
||
except Exception as e:
|
||
logger.warning("EXPLAIN 预检失败: %s", e)
|
||
result.error = str(e)[:200]
|
||
|
||
return result
|
||
|
||
|
||
async def _explain_mysql(
|
||
sql: str, instance: Instance, database: str, result: ExplainResult
|
||
) -> ExplainResult:
|
||
"""MySQL EXPLAIN 预检。"""
|
||
import aiomysql
|
||
|
||
password = decrypt_password(instance.password) if instance.password else ""
|
||
|
||
conn = await aiomysql.connect(
|
||
host=instance.host,
|
||
port=instance.port,
|
||
user=instance.username,
|
||
password=password,
|
||
db=database,
|
||
connect_timeout=5,
|
||
)
|
||
|
||
try:
|
||
sql_upper = sql.strip().upper()
|
||
|
||
# DML/DDL 类型：查询涉及的表大小
|
||
if result.sql_type in ("DML", "DDL"):
|
||
table_names = _extract_table_names(sql)
|
||
if table_names:
|
||
result.table_sizes = await _query_table_sizes_mysql(conn, database, table_names)
|
||
for table_name, row_count in result.table_sizes.items():
|
||
if row_count > LARGE_TABLE_THRESHOLD:
|
||
result.warnings.append(f"大表操作: {table_name} 约 {row_count:,} 行")
|
||
|
||
# DQL/DML: 跑 EXPLAIN
|
||
if result.sql_type in ("DQL", "DML") and not sql_upper.startswith("INSERT"):
|
||
explain_sql = f"EXPLAIN {sql}"
|
||
async with conn.cursor() as cur:
|
||
await cur.execute(explain_sql)
|
||
rows = await cur.fetchall()
|
||
columns = [desc[0] for desc in cur.description]
|
||
|
||
total_rows = 0
|
||
for row in rows:
|
||
row_dict = dict(zip(columns, row))
|
||
# MySQL EXPLAIN rows 列是预估扫描行数
|
||
scan_rows = row_dict.get("rows", 0) or 0
|
||
total_rows += scan_rows
|
||
|
||
# 检查是否全表扫描
|
||
access_type = row_dict.get("type", "")
|
||
if access_type == "ALL":
|
||
result.full_table_scan = True
|
||
result.using_index = False
|
||
|
||
# 检查 Extra 信息
|
||
extra = str(row_dict.get("Extra", ""))
|
||
if "Using filesort" in extra:
|
||
result.warnings.append("Using filesort — 需要额外排序操作")
|
||
if "Using temporary" in extra:
|
||
result.warnings.append("Using temporary — 使用临时表")
|
||
|
||
result.estimated_rows = total_rows
|
||
|
||
if total_rows > AFFECTED_ROWS_CRITICAL:
|
||
result.warnings.append(f"预估影响 {total_rows:,} 行，属于大批量操作")
|
||
elif total_rows > AFFECTED_ROWS_WARNING:
|
||
result.warnings.append(f"预估影响 {total_rows:,} 行")
|
||
|
||
# DDL ALTER: 检查是否在线 DDL
|
||
if sql_upper.startswith("ALTER TABLE"):
|
||
for table_name, row_count in result.table_sizes.items():
|
||
if row_count > LARGE_TABLE_THRESHOLD:
|
||
# MySQL 5.6+ 支持 Online DDL，但仍建议在低峰执行
|
||
result.warnings.append(
|
||
f"ALTER TABLE 在大表 {table_name}（{row_count:,} 行）上执行，建议在维护窗口操作"
|
||
)
|
||
|
||
finally:
|
||
conn.close()
|
||
|
||
return result
|
||
|
||
|
||
async def _explain_postgres(
|
||
sql: str, instance: Instance, database: str, result: ExplainResult
|
||
) -> ExplainResult:
|
||
"""PostgreSQL EXPLAIN 预检。"""
|
||
import asyncpg
|
||
|
||
password = decrypt_password(instance.password) if instance.password else ""
|
||
dsn = f"postgresql://{instance.username}:{password}@{instance.host}:{instance.port}/{database}"
|
||
|
||
conn = await asyncpg.connect(dsn, timeout=5)
|
||
|
||
try:
|
||
sql_upper = sql.strip().upper()
|
||
|
||
# DML/DDL: 查表大小
|
||
if result.sql_type in ("DML", "DDL"):
|
||
table_names = _extract_table_names(sql)
|
||
if table_names:
|
||
result.table_sizes = await _query_table_sizes_postgres(conn, table_names)
|
||
for table_name, row_count in result.table_sizes.items():
|
||
if row_count > LARGE_TABLE_THRESHOLD:
|
||
result.warnings.append(f"大表操作: {table_name} 约 {row_count:,} 行")
|
||
|
||
# DQL/DML: EXPLAIN (不带 ANALYZE，不实际执行)
|
||
if result.sql_type in ("DQL", "DML") and not sql_upper.startswith("INSERT"):
|
||
explain_sql = f"EXPLAIN (FORMAT JSON) {sql}"
|
||
rows = await conn.fetch(explain_sql)
|
||
if rows:
|
||
plan = json.loads(rows[0][0])
|
||
_parse_pg_plan(plan, result)
|
||
|
||
finally:
|
||
await conn.close()
|
||
|
||
return result
|
||
|
||
|
||
def _parse_pg_plan(plan_data: Any, result: ExplainResult) -> None:
|
||
"""递归解析 PostgreSQL EXPLAIN JSON。"""
|
||
if isinstance(plan_data, list) and plan_data:
|
||
plan_data = plan_data[0]
|
||
if isinstance(plan_data, dict):
|
||
plan = plan_data.get("Plan", plan_data)
|
||
result.estimated_rows = plan.get("Plan Rows", 0) or 0
|
||
|
||
node_type = plan.get("Node Type", "")
|
||
if "Seq Scan" in node_type:
|
||
result.full_table_scan = True
|
||
result.using_index = False
|
||
|
||
if result.estimated_rows > AFFECTED_ROWS_CRITICAL:
|
||
result.warnings.append(f"预估影响 {result.estimated_rows:,} 行，属于大批量操作")
|
||
elif result.estimated_rows > AFFECTED_ROWS_WARNING:
|
||
result.warnings.append(f"预估影响 {result.estimated_rows:,} 行")
|
||
|
||
# 递归子计划
|
||
if "Plans" in plan:
|
||
for sub_plan in plan["Plans"]:
|
||
_parse_pg_plan(sub_plan, result)
|
||
|
||
|
||
def _extract_table_names(sql: str) -> List[str]:
|
||
"""从 SQL 中提取表名（简单正则）。"""
|
||
tables = []
|
||
sql_upper = sql.upper().strip()
|
||
|
||
# UPDATE table SET ...
|
||
m = re.search(r'\bUPDATE\s+(\w+)', sql_upper)
|
||
if m:
|
||
tables.append(m.group(1).lower())
|
||
|
||
# DELETE FROM table ...
|
||
m = re.search(r'\bDELETE\s+FROM\s+(\w+)', sql_upper)
|
||
if m:
|
||
tables.append(m.group(1).lower())
|
||
|
||
# ALTER TABLE table ...
|
||
m = re.search(r'\bALTER\s+TABLE\s+(\w+)', sql_upper)
|
||
if m:
|
||
tables.append(m.group(1).lower())
|
||
|
||
# DROP TABLE table ...
|
||
m = re.search(r'\bDROP\s+TABLE\s+(?:IF\s+EXISTS\s+)?(\w+)', sql_upper)
|
||
if m:
|
||
tables.append(m.group(1).lower())
|
||
|
||
# CREATE TABLE table ...
|
||
m = re.search(r'\bCREATE\s+TABLE\s+(?:IF\s+NOT\s+EXISTS\s+)?(\w+)', sql_upper)
|
||
if m:
|
||
tables.append(m.group(1).lower())
|
||
|
||
# INSERT INTO table ...
|
||
m = re.search(r'\bINSERT\s+INTO\s+(\w+)', sql_upper)
|
||
if m:
|
||
tables.append(m.group(1).lower())
|
||
|
||
# SELECT ... FROM table (simple)
|
||
for m in re.finditer(r'\bFROM\s+(\w+)', sql_upper):
|
||
t = m.group(1).lower()
|
||
if t not in ('select', 'where', 'dual', 'information_schema'):
|
||
tables.append(t)
|
||
|
||
# JOIN table
|
||
for m in re.finditer(r'\bJOIN\s+(\w+)', sql_upper):
|
||
tables.append(m.group(1).lower())
|
||
|
||
return list(dict.fromkeys(tables)) # 去重保序
|
||
|
||
|
||
async def _query_table_sizes_mysql(conn, database: str, table_names: List[str]) -> Dict[str, int]:
|
||
"""查询 MySQL 表行数。"""
|
||
sizes = {}
|
||
if not table_names:
|
||
return sizes
|
||
|
||
placeholders = ", ".join(["%s"] * len(table_names))
|
||
query = f"""
|
||
SELECT TABLE_NAME, TABLE_ROWS
|
||
FROM information_schema.TABLES
|
||
WHERE TABLE_SCHEMA = %s AND TABLE_NAME IN ({placeholders})
|
||
"""
|
||
try:
|
||
import aiomysql
|
||
async with conn.cursor(aiomysql.DictCursor) as cur:
|
||
await cur.execute(query, [database] + [t for t in table_names])
|
||
rows = await cur.fetchall()
|
||
for row in rows:
|
||
sizes[row["TABLE_NAME"].lower()] = row.get("TABLE_ROWS", 0) or 0
|
||
except Exception as e:
|
||
logger.warning("查询表大小失败: %s", e)
|
||
|
||
return sizes
|
||
|
||
|
||
async def _query_table_sizes_postgres(conn, table_names: List[str]) -> Dict[str, int]:
|
||
"""查询 PostgreSQL 表行数。"""
|
||
sizes = {}
|
||
if not table_names:
|
||
return sizes
|
||
|
||
placeholders = ", ".join([f"'{t}'" for t in table_names])
|
||
query = f"""
|
||
SELECT relname, n_live_tup
|
||
FROM pg_stat_user_tables
|
||
WHERE relname IN ({placeholders})
|
||
"""
|
||
try:
|
||
rows = await conn.fetch(query)
|
||
for row in rows:
|
||
sizes[row["relname"].lower()] = row["n_live_tup"] or 0
|
||
except Exception as e:
|
||
logger.warning("查询PG表大小失败: %s", e)
|
||
|
||
return sizes
|