索引专题：DDL 约束 + 多轮状态追踪

本文档合并自 index_ddl_constraints.md 和 v3_force_index_index_registry_plan.md， 按"约束层 → 状态层 → 生命周期"组织，作为索引子系统的唯一参考文档。

---

第一部分：DDL 安全约束

1.1 背景

为防止 LLM 在优化过程中误修改或删除数据库原有索引，系统对 LLM 输出的 DDL 进行严格约束：采用纯白名单策略——只放行新增索引，其余一律默认拒绝。

旧版 prompt（如 opt_v2.2.txt、opt_v2.4.1.txt）曾要求模型同时生成 CREATE INDEX 和 DROP INDEX。即便当前 prompt 只要求新增索引，模型仍可能输出 DROP INDEX、ALTER ... DROP、多语句拼接等危险 DDL。因此不能信任模型输出，必须在代码层做强制过滤。

DROP INDEX 在优化流程中的职责归属：

场景	DROP INDEX 是否应出现
模型优化建议	❌ 不应该，模型只建议新增索引
测试后清理	✅ 由系统按 case 快照差集自动回滚本轮新增索引
删除生产索引	❌ 属于 DBA 职责，不在优化流程内

1.2 纯白名单过滤（唯一安全边界）

只放行「新增索引」操作，且必须是完整单条语句（正则 ^...$ 锚定）。默认拒绝：凡不严格匹配白名单的一律拒绝，无需维护「禁止操作」黑名单——黑名单天然列不全（RENAME TABLE、CREATE TABLE、GRANT、LOCK TABLES …），而白名单天然全部拦截。

唯一放行的两种语句： - CREATE INDEX / CREATE UNIQUE INDEX ... ON table (cols) - ALTER TABLE ... ADD [UNIQUE] INDEX/KEY name (cols)（仅含单个 ADD INDEX 子句，索引名必须显式给出）

被默认拒绝的典型场景：

场景	拒绝机制
DROP INDEX / ALTER ... DROP INDEX/KEY/COLUMN	不匹配任一白名单
ALTER ... RENAME/MODIFY/CHANGE	不匹配任一白名单
ANALYZE / OPTIMIZE / TRUNCATE / DROP TABLE / GRANT 等	不匹配任一白名单
多语句 ...; ...;	_has_multiple_statements() 拦截
ALTER t ADD INDEX a(x), DROP INDEX b	尾部锚定 $ 不允许第二个子句
ALTER t ADD INDEX a(x), ADD INDEX b(y)	ADD INDEX 子句计数 ≠ 1 拒绝
ALTER t ADD INDEX a(x), ADD COLUMN c	尾部锚定 $ 不允许列定义

设计要点：白名单是唯一安全边界，新增任何危险语法时无需改代码——只要它不是这两种完整单条语句，就自动被拒。

1.3 索引名自动重写

系统自动为 LLM 生成的索引名添加 llm_ 前缀，便于识别来源：

原索引名	重写后
idx_user_status	llm_idx_user_status
idx_order_date	llm_idx_order_date

长度限制： - 索引名最长 50 字符 - 超长时自动截断并追加 hash 后缀（如 llm_idx_very_long_name_..._a3f2）

1.4 SQL Hint 同步更新

当索引名被重写时，SQL 中的 FORCE INDEX / USE INDEX hint 会同步更新：

-- 原始 SQL
SELECT * FROM users FORCE INDEX(idx_user_status) WHERE status = 1

-- 重写后
SELECT * FROM users FORCE INDEX(llm_idx_user_status) WHERE status = 1

1.5 过滤与执行流程

模型输出 create_index_ddl
        ↓
_extract_create_index_ddls()     # 提取 + 规范化空白 + 去重（不做合法性判定）
        ↓
_prepare_create_index_ddls()     # 逐条白名单校验 + registry/批内去重 + 产出 hint 映射
        ↓ _parse_create_index_ddl()  # 完整单条白名单校验 + llm_ 改名（唯一安全边界）
        ↓
apply_create_index_ddls()        # 仅执行通过校验的重写后 DDL

关键点： - _parse_create_index_ddl() 对无法解析或不在白名单的 DDL 一律返回 None，_prepare_create_index_ddls() 据此跳过而非保留（历史 bug 是解析失败时误把原始 DDL 加入待执行列表，已修复）。 - hint 重写只在「索引 DDL 确认会执行」时进行（enable_index_ddl 且 create_index_ddl_complete），避免改了 hint 却没建索引。

1.6 提示词约束

提示词层面明确告知模型约束规则：

DDL 操作约束（强制）
create_index_ddl 数组仅允许以下操作：
规则1：只允许 CREATE INDEX 或 CREATE UNIQUE INDEX 语句。
规则2：禁止 DROP INDEX、ALTER TABLE DROP INDEX/KEY（不允许删除索引）。
规则3：禁止 ALTER TABLE ADD INDEX/KEY（请使用 CREATE INDEX）。
规则4：禁止 ALTER TABLE RENAME INDEX/KEY（不允许重命名索引）。
规则5：禁止 ANALYZE TABLE、OPTIMIZE TABLE 等表维护操作。

---

第二部分：多轮索引状态追踪

2.1 背景问题

当前 V3 多轮 SQL 优化流程中，存在一种跨轮索引状态不一致问题：

1. 第二轮 LLM 输出了 CREATE INDEX，系统执行该 DDL。
2. 第二轮执行完成后，系统按快照差集回滚本轮新增索引。
3. 第三轮 LLM 认为第二轮索引是正确的，因此没有重新输出 create_index_ddl。
4. 第三轮 SQL 又加入了 FORCE INDEX(idx_xxx) 或继续保留上一轮 hint。
5. 当前执行环境中该索引已经不存在，最终数据库报错。

本质问题是：

LLM 认为"上一轮索引策略有效"
        !=
当前数据库执行环境中"该索引真实存在"

因此，系统不能把"上一轮索引有效"只保存在模型记忆里，却在数据库执行环境中每轮回滚。推荐方案是让同一个 case 内的索引环境累积保留，并用系统解析结果记录"本轮创建了什么、SQL hint 指向什么、EXPLAIN 实际用了什么"。如果 hint 指向的索引不存在，不做额外拦截，让数据库返回错误并写入 exec_error，再反馈给下一轮修正。

2.2 当前代码现状

当前关键链路如下：

BuildPrompt
  -> CallLLM
  -> ParseLLM
  -> Persist
  -> ExecuteSQL
  -> PlanReview
  -> Feedback
  -> StopPolicy

相关文件：

• core/llm_opt_v3/v3_nodes.py
• core/tdsql_client/sql_exe.py

当前 node_execute 的主要行为：

create_index_ddls = _extract_create_index_ddls(parsed)

if enable_index_ddl and create_index_ddls and create_index_ddl_complete:
    before_snapshot = snapshot_non_primary_indexes(client)
    ddl_result = apply_create_index_ddls(client, create_index_ddls)

exe_opt_sql(...)

finally:
    if enable_index_ddl and before_snapshot is not None:
        drop_indexes_by_snapshot_diff(client, before_snapshot)

这个逻辑只会执行"当前轮 last_parsed.create_index_ddl 里的 DDL"。如果第三轮没有重新输出 DDL，系统不会自动创建第二轮索引。

同时，drop_indexes_by_snapshot_diff 会回滚本轮新增索引，所以第二轮创建的索引不会自然留到第三轮。

2.3 设计目标

目标不是给 FORCE INDEX / USE INDEX 增加一层复杂预校验，而是把多轮优化中的索引事实记录清楚，并把数据库执行环境控制在 case 边界内。

需要满足以下约束：

• 同一个 case 内不按轮回滚索引，前面轮次创建成功的索引可以被后续轮次自然使用。
• 不同 case 之间必须隔离，case 结束后清理本 case 创建过的索引，避免污染下一个 case。
• FORCE INDEX / USE INDEX 只作为 SQL hint 事实记录；系统解析为 hint_index_keys，不做复杂继承重建，也不做 catalog / registry 预校验。
• 如果 hint 引用索引不存在，由数据库执行或 EXPLAIN 返回错误，系统将错误写入 exec_error 并反馈下一轮。
• 每轮 JSON 只记录三个系统回填字段：本轮成功创建的 created_index_keys、本轮 SQL 显式引用的 hint_index_keys、本轮 EXPLAIN 实际命中的 used_index_keys。
• 索引 DDL、创建轮次、状态等详情由 case 级 index_registry 保存，不在每轮 JSON 中重复落盘。

2.4 总体方案

引入 case 级索引注册表 index_registry，记录当前 case 内 LLM 建议并成功应用过的索引 DDL。

每一轮执行时，系统只做三件事：

1. 执行当前轮 LLM 输出的 create_index_ddl，成功后写入 index_registry 并回填 created_index_keys。
2. 执行 SQL 和 EXPLAIN，解析实际命中的索引，回填 used_index_keys。
3. 如果 SQL 中有 FORCE INDEX / USE INDEX，系统只解析并回填 hint_index_keys，不做继承重建或复杂预校验。

执行流程调整为：

解析当前轮 create_index_ddl
        |
        v
解析 optimized_sql 中的 FORCE INDEX / USE INDEX
        |
        v
生成 hint_index_keys
        |
        v
执行当前轮 create_index_ddl
        |
        v
创建成功的索引写入 index_registry，并生成 created_index_keys
        |
        v
执行 optimized_sql
        |
        v
执行 EXPLAIN，解析 used_index_keys
        |
        v
将 created_index_keys / hint_index_keys / used_index_keys 回填到本轮 JSON
        |
        v
case 结束后按 index_registry 清理本 case 创建的索引

2.5 状态结构

在 V3State 中新增字段：

class V3State(TypedDict, total=False):
    ...
    case_initial_index_snapshot: Optional[set[tuple[str, str]]]
    index_registry: Dict[str, Dict[str, Any]]
    index_cleanup_done: bool
    index_cleanup_payload: Dict[str, Any]

index_registry 推荐结构：

{
  "products.idx_products_discontinued_list_price_cover": {
    "table": "products",
    "index_name": "idx_products_discontinued_list_price_cover",
    "ddl": "CREATE INDEX idx_products_discontinued_list_price_cover ON products(discontinued, list_price, id);",
    "created_round": 2,
    "status": "applied"
  }
}

key 统一使用：

f"{table_name}.{index_name}".lower()

字段含义：

字段	含义
case_initial_index_snapshot	当前 case 开始前的数据库索引快照
index_registry	当前 case 中系统成功创建过的索引
index_cleanup_done	是否已经执行过 case 级清理
index_cleanup_payload	清理成功、失败、残留检查等信息

2.6 工具函数

2.6.1 解析 CREATE INDEX DDL

用于从 LLM 输出的 DDL 中提取表名和索引名。

def _parse_create_index_ddl(ddl: str) -> Optional[Dict[str, str]]:
    patterns = [
        re.compile(
            r"\bcreate\s+(?:unique\s+)?index\s+`?([A-Za-z0-9_]+)`?\s+on\s+`?([A-Za-z0-9_]+)`?",
            re.IGNORECASE,
        ),
        re.compile(
            r"\balter\s+table\s+`?([A-Za-z0-9_]+)`?\s+add\s+(?:unique\s+)?(?:index|key)\s+`?([A-Za-z0-9_]+)`?",
            re.IGNORECASE,
        ),
    ]

    text = re.sub(r"\s+", " ", ddl.strip())
    if not text:
        return None

    m = patterns[0].search(text)
    if m:
        return {
            "index_name": m.group(1),
            "table": m.group(2),
            "ddl": text if text.endswith(";") else text + ";",
        }

    m = patterns[1].search(text)
    if m:
        return {
            "table": m.group(1),
            "index_name": m.group(2),
            "ddl": text if text.endswith(";") else text + ";",
        }

    return None

2.6.2 解析 SQL 中的索引 hint

用于从 optimized_sql 中提取 FORCE INDEX / USE INDEX 引用。

def _extract_index_hint_refs(sql: str) -> List[Dict[str, str]]:
    refs: List[Dict[str, str]] = []
    if not sql:
        return refs

    pattern = re.compile(
        r"\b(?:from|join)\s+`?([A-Za-z0-9_]+)`?"
        r"(?:\s+(?:as\s+)?`?[A-Za-z0-9_]+`?)?"
        r"\s+(force|use)\s+index\s*(?:for\s+\w+)?\s*\(([^)]*)\)",
        re.IGNORECASE,
    )

    seen = set()
    for m in pattern.finditer(sql):
        table = m.group(1)
        hint_type = m.group(2).upper()
        raw_indexes = m.group(3)
        for idx in raw_indexes.split(","):
            index_name = idx.strip().strip("`").strip()
            if not index_name:
                continue
            key = f"{table}.{index_name}".lower()
            if key in seen:
                continue
            seen.add(key)
            refs.append({
                "table": table,
                "index_name": index_name,
                "hint_type": hint_type,
                "key": key,
            })

    return refs

2.6.3 解析 EXPLAIN 中的实际命中索引

用于把数据库执行计划中的 key 字段转成统一的 used_index_keys。

第一版只需要解析"实际使用的索引"，不要把 possible_keys 当作命中结果。

def extract_used_index_keys_from_explain(
    explain_rows: list[dict[str, Any]],
    table_alias_map: dict[str, str] | None = None,
) -> list[str]:
    used: list[str] = []
    seen: set[str] = set()

    for row in explain_rows or []:
        table = str(row.get("table") or "").strip()
        index_name = str(row.get("key") or "").strip()
        if not index_name or index_name.upper() == "NULL":
            continue

        real_table = (table_alias_map or {}).get(table, table)
        key = f"{real_table}.{index_name}".lower() if real_table else f".{index_name}".lower()
        if key in seen:
            continue
        seen.add(key)
        used.append(key)

    return used

2.7 node_execute 改造

2.7.1 解析当前轮 DDL 与 hint

在 node_execute 中，先解析当前轮 DDL 和 SQL hint。这里先不写 registry，只有 DDL 实际创建成功后才登记：

current_ddls = _extract_create_index_ddls(parsed if isinstance(parsed, dict) else {})
optimized_sql = (parsed.get("optimized_sql") or "").strip() if isinstance(parsed, dict) else ""
hint_refs = _extract_index_hint_refs(optimized_sql)
hint_index_keys = [ref["key"] for ref in hint_refs]

index_registry = dict(state.get("index_registry") or {})
current_ddl_specs = {}

for ddl in current_ddls:
    spec = _parse_create_index_ddl(ddl)
    if not spec:
        continue
    key = f"{spec['table']}.{spec['index_name']}".lower()
    current_ddl_specs[key] = spec

2.7.2 解析 FORCE / USE INDEX 为 hint_index_keys

只解析并记录 hint 中出现的索引 key：

hint_index_keys = [ref["key"] for ref in hint_refs]

如果 hint 引用的索引不存在，数据库会在执行 SQL 或 EXPLAIN 时返回错误。系统把错误写入 exec_error，并把 hint_index_keys 保留在本轮 JSON 中，供下一轮模型修正。

2.7.3 执行当前轮 DDL 并生成 created_index_keys

ddl_result = apply_create_index_ddls(client, current_ddls)

created_index_keys = []
for ddl in ddl_result.get("applied", []):
    spec = _parse_create_index_ddl(ddl)
    if not spec:
        continue
    key = f"{spec['table']}.{spec['index_name']}".lower()
    created_index_keys.append(key)
    index_registry[key] = {
        "table": spec["table"],
        "index_name": spec["index_name"],
        "ddl": spec["ddl"],
        "created_round": state["round"],
        "status": "applied",
    }

exec_payload 增加：

exec_payload.update({
    "created_index_keys": created_index_keys,
    "hint_index_keys": hint_index_keys,
})

2.8 registry 状态更新

执行 DDL 后，只把创建成功的索引写回 registry：

def _register_applied_indexes(
    index_registry: Dict[str, Dict[str, Any]],
    ddl_result: Dict[str, Any],
    round_idx: int,
) -> Dict[str, Dict[str, Any]]:
    updated = dict(index_registry)

    for ddl in ddl_result.get("applied", []) or []:
        spec = _parse_create_index_ddl(ddl)
        if not spec:
            continue
        key = f"{spec['table']}.{spec['index_name']}".lower()
        item = dict(updated.get(key) or {})
        item.update({
            "table": spec["table"],
            "index_name": spec["index_name"],
            "ddl": spec["ddl"],
            "created_round": round_idx,
            "status": "applied",
        })
        updated[key] = item

    return updated

2.9 每轮 index_trace 格式

每轮 JSON 只记录三个执行事实：

{
  "index_trace": {
    "created_index_keys": ["orders.idx_orders_customer_id"],
    "hint_index_keys": ["orders.idx_orders_order_date_id"],
    "used_index_keys": ["orders.idx_orders_order_date_id", "customers.PRIMARY"]
  }
}

字段	生成方	含义
created_index_keys	系统	本轮 create_index_ddl 实际创建成功的索引 key
hint_index_keys	系统	本轮 SQL 中 FORCE INDEX / USE INDEX 显式引用的索引 key
used_index_keys	系统	本轮 EXPLAIN 实际命中的索引 key

key 统一格式：<table_name_lower>.<index_name_lower>

不要在每轮 JSON 中保存"命中的索引里哪些是本 case 创建的""创建但未命中的索引有哪些""最终有效 DDL 是什么"等派生信息。这些都可由 index_trace + index_registry 重新计算。

---

第三部分：索引生命周期（回滚与清理）

3.1 回滚策略

推荐把回滚粒度从 round 级调整为 case 级：

同一个 case 内：不按轮回滚索引
不同 case 之间：必须清理本 case 新增索引

case_start:
  记录 case_initial_index_snapshot
  初始化 index_registry = {}

case rounds:
  每轮执行 create_index_ddl
  创建成功的索引写入 index_registry
  每轮不回滚
  每轮解析 EXPLAIN，回填 created_index_keys / used_index_keys

case_end:
  优先按 index_registry 精确删除本 case 创建过的索引
  再用 case_initial_index_snapshot 做残留检查

3.2 为什么不用每轮回滚

当前问题的核心就是每轮回滚导致：

第二轮创建索引
第二轮结束后回滚
第三轮 SQL 没有重新输出索引
第三轮无法使用第二轮索引

如果要让第三轮能自然命中第二轮索引，就必须让同一个 case 内的索引跨轮保留。

3.3 为什么不能永久不回滚

完全不回滚会污染后续 case：

case A 第二轮创建 idx_orders_order_date_id
case A 结束后不删除
case B 开始时也查询 orders
case B 的执行计划可能自然命中 case A 留下的索引

3.4 清理方式：优先 registry 精确清理

差集删除有并发风险（可能误删其他任务创建的索引），推荐主清理方式改为按 index_registry 精确删除：

def drop_indexes_by_registry(client: TDSQLClient, index_registry: Dict[str, Dict[str, Any]]) -> tuple[int, int]:
    success_count = 0
    fail_count = 0

    for item in index_registry.values():
        table = str(item.get("table") or "").strip()
        index_name = str(item.get("index_name") or "").strip()
        if not table or not index_name:
            continue

        safe_table = table.replace("`", "``")
        safe_idx = index_name.replace("`", "``")
        ddl = f"ALTER TABLE `{safe_table}` DROP INDEX `{safe_idx}`;"

        try:
            client.execute_query(ddl)
            success_count += 1
        except Exception:
            fail_count += 1

    return success_count, fail_count

优点：只删除当前 case 中系统成功创建并登记过的索引，不会误删其他并发任务创建的索引。

3.5 快照的职责：兜底检查，不默认删除未知差集

case 结束后，先按 registry 清理，再做残留检查：

drop_indexes_by_registry(client, index_registry)

after_snapshot = snapshot_non_primary_indexes(client)
remaining_diff = after_snapshot - case_initial_index_snapshot

• 如果 remaining_diff 中的索引属于 index_registry：说明 registry 清理失败，记录 cleanup_failed。
• 如果 remaining_diff 中的索引不属于 index_registry：说明可能是其他任务创建的索引，记录 warning，不自动删除。

3.6 异常兜底清理

case 正常结束时可以在 node_stop 中清理，但如果流程中间抛异常，node_stop 可能执行不到。

还需要在 run_batch 的每个 folder 外层加 try/finally 兜底：

for idx, folder in enumerate(folders, 1):
    final_state = None
    case_initial_index_snapshot = None
    index_registry = {}

    try:
        client = TDSQLClient(db_section=db_section)
        case_initial_index_snapshot = snapshot_non_primary_indexes(client)

        state["case_initial_index_snapshot"] = case_initial_index_snapshot
        state["index_registry"] = index_registry
        state["index_cleanup_done"] = False

        final_state = graph.invoke(state)

    finally:
        cleanup_done = bool(final_state and final_state.get("index_cleanup_done"))
        if not cleanup_done:
            client = TDSQLClient(db_section=db_section)
            registry = (final_state or state).get("index_registry") or index_registry
            drop_indexes_by_registry(client, registry)

---

第四部分：Prompt 与输出协议

4.1 当前方案

LLM 仍然只需要输出原有的：

{
  "optimized_sql": "SELECT ...",
  "create_index_ddl": []
}

index_trace.created_index_keys、index_trace.hint_index_keys 和 index_trace.used_index_keys 由系统执行后回填，不进入 LLM 输出 schema。

Prompt 中只需要补充以下硬规则：

索引输出规则：
1. 如果本轮建议新增索引，必须继续写入 create_index_ddl。
2. 如果本轮没有新增索引，create_index_ddl 保持 []。
3. 如果 optimized_sql 中包含 FORCE INDEX / USE INDEX，可以保留；系统会解析并写入 index_trace.hint_index_keys。
4. 不要求模型判断历史索引是否实际命中；实际命中由系统解析 EXPLAIN 后写入 index_trace.used_index_keys。
5. 不要求模型输出 hint_index_keys、used_index_keys、派生 DDL 或 index_policy。
6. 如果 hint 引用索引不存在，数据库执行错误会写入 exec_error，并反馈下一轮。

4.2 长期方案：索引命名权收归系统

当前实现（过渡方案）： 系统自动为 LLM 生成的索引名添加 llm_ 前缀（详见第一部分 1.3、1.4）。

长期方案： 把 LLM 输出从 DDL 字符串改成结构化建议：

{
  "index_recommendations": [
    {
      "table": "products",
      "columns": ["discontinued", "list_price", "id"],
      "kind": "btree",
      "purpose": "filter_exists_products"
    }
  ]
}

系统生成稳定索引名：

llm_idx_products_discontinued_list_price_id_a13f

DDL 和 hint 都由同一个 registry 生成，避免第二轮和第三轮索引名不一致。

---

第五部分：实现位置

文件	功能
core/llm_opt_v3/index_ddl.py	DDL 过滤、索引名重写、hint 同步、registry 管理
core/llm_opt_v3/v3_nodes.py	node_execute 改造、case 级清理
core/tdsql_client/sql_exe.py	DDL 执行、EXPLAIN 执行、错误捕获
core/llm_opt_v3/prompts/opt_v3.2.0.txt	提示词约束
core/llm_opt_v3/prompts/opt_v3.1.0.txt	提示词约束

---

第六部分：测试覆盖

单测文件：tests/test_index_ddl_rewrite.py

DDL 约束测试

• 正常索引名加 llm_ 前缀
• 已有 llm_ 前缀不再重复添加
• 超长索引名截断
• CREATE INDEX / CREATE UNIQUE INDEX / 带 schema 限定 解析重写
• ALTER TABLE ADD INDEX/KEY（含 UNIQUE、前缀索引）放行并重写
• 非白名单语句（DROP/RENAME/MODIFY/ANALYZE/OPTIMIZE/TRUNCATE/DROP TABLE/CREATE TABLE/GRANT …）默认拒绝
• 多语句拼接拒绝
• ALTER 夹带 DROP / 多个 ADD INDEX / ADD COLUMN 拒绝
• FORCE/USE INDEX 单/多索引替换
• 多表查询分别替换
• 索引不在映射中保持不变
• registry 去重 + 批内去重

多轮状态追踪测试

• 第三轮强制引用第二轮索引（case 内累计索引环境 + hint 解析）
• SQL 引用未知 hint 索引（系统不提前拦截，错误来自数据库执行）
• 当前轮新 DDL 与 FORCE INDEX 匹配
• 第三轮无 hint 但自然命中第二轮索引
• 第三轮创建两个索引但只命中一个（created_index_keys 与 used_index_keys 分开）
• 每轮 JSON 只保存最小 index_trace（三个 key）
• case 结束按 registry 精确清理
• 异常中断时 finally 兜底清理
• 快照差集只做未知残留告警

---

第七部分：分阶段实施计划

阶段一：最小 index_trace 回填

1. 在 V3State 中新增 index_registry。
2. 在 node_load_inputs 初始化 index_registry。
3. 新增 _parse_create_index_ddl，从当前轮 create_index_ddl 解析索引 key。
4. 新增 _extract_index_hint_refs，从 optimized_sql 解析 hint_index_keys。
5. 改造 node_execute，当前轮 DDL 创建成功后回填 created_index_keys 并写入 index_registry。
6. 执行 SQL / EXPLAIN 后解析 used_index_keys。
7. 将三个 key 回填到本轮 JSON 的 index_trace。
8. 捕获数据库执行或 EXPLAIN 错误，写入 exec_error。

阶段二：case 累计索引环境

1. 同一个 case 内不按轮执行 drop_indexes_by_snapshot_diff。
2. case 开始前记录 case_initial_index_snapshot。
3. 每轮成功创建的索引写入 case 级 index_registry。
4. 后续轮次直接在累计后的数据库索引环境中执行。
5. case 正常结束时按 index_registry 精确清理本 case 新增索引。
6. 清理后用 case_initial_index_snapshot 做残留检查，未知差集只告警不自动删除。
7. run_batch 外层 finally 做兜底清理，避免异常导致索引残留。

阶段三：反馈与 prompt 增强

1. exec_payload 增加 created_index_keys、hint_index_keys、used_index_keys、exec_error、index_registry。
2. build_plan_review 增加索引状态摘要。
3. build_feedback 把本轮索引 key 传给下一轮。
4. 修改 generation / review prompt，要求模型关注索引创建/hint/命中情况。

阶段四：索引命名权收归系统

详见第四部分 4.2 长期方案。

---

第八部分：验收标准

• 同一个 case 内，前面轮次创建成功的索引可以继续留在执行环境中，后续轮次可由优化器自然选择。
• 第三轮即使没有重新输出 create_index_ddl，只要 EXPLAIN 命中第二轮创建的索引，本轮 JSON 的 used_index_keys 能记录该 key。
• 每轮 JSON 的 index_trace 只持久化 created_index_keys、hint_index_keys、used_index_keys。
• created_index_keys 与 used_index_keys 分开记录，不能把"本轮创建成功"等同于"本轮执行计划实际命中"。
• SQL 中所有 FORCE INDEX / USE INDEX 引用都能被解析并写入 hint_index_keys。
• hint 引用索引不存在时，系统不做额外拦截；数据库执行或 EXPLAIN 返回的错误写入 exec_error。
• case 级 index_registry 保存索引 DDL、创建轮次、状态等详细信息。
• 每轮 JSON 不写入派生字段。
• EXPLAIN 解析结果能记录本轮实际命中的 used_index_keys。
• 同一个 case 内不按轮回滚索引，但 case 结束后必须按 index_registry 清理本 case 新增索引。
• case 初始快照只用于残留检查；未知差集只告警，不默认删除。
• 单元测试覆盖新增逻辑的正常路径和失败路径。