大语言模型驱动的 Text2SQL 技术显著降低了业务人员探查数据的技术门槛。但在企业级数据架构中，直接让大模型生成的 SQL 穿透到只读从库甚至分析实例上执行，等同于将数据库的可用性完全赌在模型的生成概率上。
大语言模型在生成 SQL 时存在天然的不稳定性：缺少强制主键或索引字段的过滤条件、误用隐式类型转换导致 B+ 树索引失效、在多表关联中遗漏连接条件诱发上亿条记录的笛卡尔积、或者在没有分页的场景下使用深层嵌套子查询。一旦恶性 SQL 在高并发或大表场景下触发全表扫描（Full Table Scan），不仅会瞬间打满只读节点的磁盘 IOPS，还会冲刷掉整个 InnoDB Buffer Pool 中的热点缓存页，引发全局只读副本的雪崩。
为解决这一问题，我们在查询网关层引入了“执行计划预检沙箱”机制。所有大模型生成的 SQL 必须先在专用的只读测试实例上执行EXPLAIN FORMAT=JSON。只有通过代价模型规则校验的语句，才被允许下发至生产环境。
预检网关架构与拦截规则
校验流水线部署在模型调度服务与数据库连接池之间。预检实例通常挂载与生产库结构一致、但仅包含采样数据（或由物理备份每日脱敏还原）的只读节点。
核心拦截逻辑聚焦于 MySQLEXPLAIN FORMAT=JSON暴露的几个核心元数据指标：
- 扫描类型拦截 (
access_type)：若access_type为ALL（全表扫描）或index（全索引扫描），且预估扫描行数rows_examined_per_scan超过安全阈值（如 50,000 行），一票否决。 - 连接类型与笛卡尔积拦截：在多表关联时，若出现
type为ALL且没有外键/连接键约束匹配，或者filtered比例低于安全边界，判定为笛卡尔积风险。 - 临时表与文件排序开销 (
using_temporary_table,using_filesort)：大表上同时触发磁盘临时表（Disk Temporary Table）与外部文件排序的复杂查询直接打回，要求模型重写或强制增加索引提示。 - 隐式类型转换导致的索引失效：当模型在
WHERE条件中对数值型主键传入字符串（或反之），导致possible_keys有值但实际key为NULL时，强制熔断并提取修正提示反馈给大模型。
预检拦截网关的工程实现
以下为基于 Python 实现的轻量级执行计划校验器核心代码。代码直接解析 MySQL 8.0 规范的 JSON 执行计划树，支持递归遍历嵌套循环连接（Nested Loop Join）与物化子查询：
import json import logging from typing import Dict, Any, Tuple, List import pymysql logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s") logger = logging.getLogger("ExplainGuard") class ExplainInterceptor: def __init__(self, host: str, port: int, user: str, password: str, db: str): self.conn_params = { "host": host, "port": port, "user": user, "password": password, "database": db, "cursorclass": pymysql.cursors.DictCursor, "autocommit": True } # 硬性红线规则配置 self.MAX_EXAMINED_ROWS_LIMIT = 50000 self.MAX_TOTAL_COST_LIMIT = 10000.0 def get_connection(self): return pymysql.connect(**self.conn_params) def validate_sql(self, sql_query: str) -> Tuple[bool, str, Dict[str, Any]]: """ 在只读测试实例中获取执行计划并进行深度静态评估 """ # 强制设置只读上下文与最长执行超时，防止 EXPLAIN 本身被恶性语句挂起 explain_sql = f"/*+ MAX_EXECUTION_TIME(3000) */ EXPLAIN FORMAT=JSON {sql_query.strip().rstrip(';')}" try: with self.get_connection() as conn: with conn.cursor() as cursor: # 先行设置事务只读 cursor.execute("SET TRANSACTION READ ONLY;") cursor.execute(explain_sql) result = cursor.fetchone() if not result or "EXPLAIN" not in result: return False, "未能获取到有效的执行计划输出", {} plan_data = json.loads(result["EXPLAIN"]) except pymysql.MySQLError as err: logger.error("EXPLAIN 执行异常: %s, 原始 SQL: %s", err, sql_query) return False, f"SQL语法校验或执行计划提取失败: {str(err)}", {} # 遍历解析查询树 violations = [] metrics = {"total_cost": 0.0, "max_rows_scanned": 0, "has_filesort": False} query_block = plan_data.get("query_block", {}) self._analyze_query_block(query_block, violations, metrics) if violations: violation_summary = " | ".join(violations) logger.warning("SQL 拦截拦截触发: %s", violation_summary) return False, f"执行计划未通过红线检查: {violation_summary}", metrics return True, "执行计划校验通过", metrics def _analyze_query_block(self, block: Dict[str, Any], violations: List[str], metrics: Dict[str, Any]): cost_info = block.get("cost_info", {}) query_cost = float(cost_info.get("query_cost", 0.0)) metrics["total_cost"] = max(metrics["total_cost"], query_cost) if query_cost > self.MAX_TOTAL_COST_LIMIT: violations.append(f"单查询预估总代价值超标 (Cost: {query_cost} > {self.MAX_TOTAL_COST_LIMIT})") # 检查是否包含文件排序 if block.get("ordering_operation", {}).get("using_filesort", False): metrics["has_filesort"] = True # 检查嵌套循环表扫描 if "nested_loop" in block: for item in block["nested_loop"]: table_info = item.get("table", {}) self._check_table_scan(table_info, violations, metrics) elif "table" in block: self._check_table_scan(block["table"], violations, metrics) def _check_table_scan(self, table: Dict[str, Any], violations: List[str], metrics: Dict[str, Any]): table_name = table.get("table_name", "UNKNOWN") access_type = table.get("access_type", "") rows_examined = int(table.get("rows_examined_per_scan", 0)) metrics["max_rows_scanned"] = max(metrics["max_rows_scanned"], rows_examined) # 核心红线 1: 全表扫描行数超限 if access_type in ("ALL", "index") and rows_examined > self.MAX_EXAMINED_ROWS_LIMIT: violations.append( f"表 [{table_name}] 触发全表/全索引扫描，扫描行数: {rows_examined}，超出阈值: {self.MAX_EXAMINED_ROWS_LIMIT}" ) # 核心红线 2: 索引覆盖缺失下的临时表混用文件排序 if table.get("using_temporary_table", False) and table.get("using_filesort", False): violations.append(f"表 [{table_name}] 同时触发内存临时表与外部文件排序")实战测试验证
我们在只读测试机上部署了带有 200 万条记录的订单明细表order_items。模拟大模型在接收业务问题“查询所有包含手机配件的订单且按创建时间排序”时生成的劣质 SQL 与优化后 SQL。
劣质 SQL 的拦截验证
大模型初次生成的 SQL 如下：
SELECT order_id, product_name, price FROM order_items WHERE SUBSTRING(product_name, 1, 4) = '配件' ORDER BY create_time DESC;执行拦截器检验后，解析到的执行计划输出显示：
access_type:ALLrows_examined_per_scan:1985420using_filesort:truequery_cost:201452.40
拦截器立即打断执行，控制台输出：
WARNING - SQL 拦截拦截触发: 表 [order_items] 触发全表/全索引扫描，扫描行数: 1985420，超出阈值: 50000 | 单查询预估总代价值超标 (Cost: 201452.4 > 10000.0)网关将该拦截摘要封装为上下文直接回传给大模型：“检测到 product_name 上存在函数导致索引失效，且触发近 200 万行的全表扫描，请去除字段上的函数包装改用前缀模糊匹配或关联索引”。
模型接收反馈后自我纠偏重写出合规 SQL：
SELECT order_id, product_name, price FROM order_items WHERE product_name LIKE '配件%' AND create_time >= '2026-10-01 00:00:00' ORDER BY create_time DESC LIMIT 100;再次运行预检，access_type变为range，扫描行数降低至 1,240 行，顺利放行流入只读执行集群。
生产落地的权衡与避坑要点
- 统计信息漂移导致执行计划失真：预检实例的数据量若是小样本采样，其
mysql.innodb_table_stats与生产库的主键基数（Cardinality）存在巨大差异，容易导致测试机走索引而生产库选错走全表。生产规范中必须每晚定时同步生产库的持久化统计信息表至预检实例，或者通过ANALYZE TABLE固定采样率。 - 只读测试实例的强行熔断：即使是
EXPLAIN，在 MySQL 解析特定包含多层子查询的视图时，仍然可能会触发子查询的物化（Materialization）并实际运行部分算子。因此在预检连接建立时，必须通过SET SESSION max_execution_time = 3000;注入超时熔断，防止劣质执行计划反向把预检实例夯死。 - 闭环反馈机制建设：拦截不仅要做到防御，更要把拦截明细指标（命中哪条红线、缺失哪类索引）结构化注入给大模型的下次 Prompt。从实测统计来看，配备了执行计划阻断与反馈循环的系统，模型重写后二次通过率高达 91%，真正将不可靠的概率输出约束为确定性的生产级 SQL。
