大语言模型驱动的 Text2SQL 技术显著降低了业务人员探查数据的技术门槛。但在企业级数据架构中直接让大模型生成的 SQL 穿透到只读从库甚至分析实例上执行等同于将数据库的可用性完全赌在模型的生成概率上。大语言模型在生成 SQL 时存在天然的不稳定性缺少强制主键或索引字段的过滤条件、误用隐式类型转换导致 B 树索引失效、在多表关联中遗漏连接条件诱发上亿条记录的笛卡尔积、或者在没有分页的场景下使用深层嵌套子查询。一旦恶性 SQL 在高并发或大表场景下触发全表扫描Full Table Scan不仅会瞬间打满只读节点的磁盘 IOPS还会冲刷掉整个 InnoDB Buffer Pool 中的热点缓存页引发全局只读副本的雪崩。为解决这一问题我们在查询网关层引入了“执行计划预检沙箱”机制。所有大模型生成的 SQL 必须先在专用的只读测试实例上执行EXPLAIN FORMATJSON。只有通过代价模型规则校验的语句才被允许下发至生产环境。预检网关架构与拦截规则校验流水线部署在模型调度服务与数据库连接池之间。预检实例通常挂载与生产库结构一致、但仅包含采样数据或由物理备份每日脱敏还原的只读节点。核心拦截逻辑聚焦于 MySQLEXPLAIN FORMATJSON暴露的几个核心元数据指标扫描类型拦截 (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(levellogging.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 FORMATJSON {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, fSQL语法校验或执行计划提取失败: {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 万行的全表扫描请去除字段上的函数包装改用前缀模糊匹配或关联索引”。模型接收反馈后自我纠偏重写出合规 SQLSELECT 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。