从 Apache Phoenix 迁移到国产库，10万行 SQL 全是 HBase 方言！我用 AI Agent + AST 变异引擎，72小时完成“不可能的迁移“，DBA 看完当场哭了！

墨夶

20人浏览 · 2026-07-08 10:30:00

墨夶 · 2026-07-08 10:30:00 发布

一、案发现场：为什么 Phoenix SQL 迁移是"地狱难度"？

物流公司，核心运单追踪系统基于 HBase + Apache Phoenix 构建。Phoenix 是 HBase 的 SQL 皮肤，让开发者能用 SQL 语法操作 HBase 的宽表。几年下来，系统积累了：

3200+ 条业务 SQL（嵌在 Java 代码、MyBatis XML、存储过程里）
450+ 张 Phoenix 表定义（包含 Salting、Column Family、Immutable Rows 等 HBase 特有概念）
80+ 个 Phoenix 二级索引（包括 Local Index、Global Index、Covered Index）
30+ 个 UPSERT VALUES 批量写入逻辑

现在信创要求：HBase 下线，迁移到国产数据库。

DBA 团队看了一眼 Phoenix 的 SQL，当场崩溃：

– Phoenix 特有语法，任何国产库都不认识！

– 1. Salting 建表（HBase 数据分散策略）
CREATE TABLE IF NOT EXISTS LOGISTICS.ORDER_TRACK (
track_id VARCHAR NOT NULL,
order_id VARCHAR NOT NULL,
event_time TIMESTAMP NOT NULL,
status VARCHAR,
location VARCHAR,
operator VARCHAR,
detail VARCHAR
CONSTRAINT pk PRIMARY KEY (track_id, order_id, event_time)
) SALT_BUCKETS=16,
IMMUTABLE_ROWS=true,
COMPRESSION=‘SNAPPY’,
COLUMN_ENCODED_BYTES=0;

– 2. 列族声明（Phoenix 独有的列族语法）
CREATE TABLE LOGISTICS.SHIPMENT_DETAIL (
shipment_id VARCHAR NOT NULL,
– 默认列族的列
origin VARCHAR,
destination VARCHAR,
– info 列族的列（Phoenix 用列族.列名的语法）
info.sender_name VARCHAR,
info.sender_phone VARCHAR,
info.receiver_name VARCHAR,
info.receiver_phone VARCHAR,
– cargo 列族的列
cargo.weight DOUBLE,
cargo.volume DOUBLE,
cargo.items INTEGER
CONSTRAINT pk PRIMARY KEY (shipment_id)
);

– 3. Phoenix UPSERT（不是标准 SQL 的 INSERT，是 PUT 语义！）
UPSERT INTO LOGISTICS.ORDER_TRACK
(track_id, order_id, event_time, status, location)
VALUES (‘TRK001’, ‘ORD20260101’, CURRENT_TIMESTAMP(), ‘IN_TRANSIT’, ‘上海转运中心’);

– 4. Phoenix UPSERT … SELECT（批量数据搬运）
UPSERT INTO LOGISTICS.ORDER_ARCHIVE
SELECT * FROM LOGISTICS.ORDER_TRACK
WHERE event_time < CURRENT_DATE() - 90;

– 5. Phoenix 全局索引（底层是另一张 HBase 表）
CREATE INDEX IDX_ORDER_STATUS ON LOGISTICS.ORDER_TRACK (status, event_time DESC)
INCLUDE (location, operator);

– 6. Phoenix 本地索引（与数据表共处一个 Region）
CREATE LOCAL INDEX IDX_SHIPMENT_ORIGIN ON LOGISTICS.SHIPMENT_DETAIL (origin);

– 7. Phoenix 特有函数
SELECT
track_id,
order_id,
– ARRAY 类型和操作（Phoenix 支持数组列）
ARRAY_ELEM(status_history, 1) AS first_status,
ARRAY_LENGTH(status_history) AS history_count,
– 字节操作函数
TO_HEX(track_id) AS hex_id,
– RowKey 范围扫描提示
/*+ RANGE_SCAN */
event_time
FROM LOGISTICS.ORDER_TRACK
WHERE track_id >= ‘TRK000’ AND track_id < ‘TRK999’
AND status IN (‘DELIVERED’, ‘RETURNED’);

– 8. Phoenix 序列
CREATE SEQUENCE LOGISTICS.ORDER_SEQ START WITH 1 INCREMENT BY 1;
SELECT NEXT VALUE FOR LOGISTICS.ORDER_SEQ FROM SYSTEM.SEQUENCE;

– 9. Phoenix Hint（查询优化提示，与 Oracle Hint 不同）
SELECT /*+ INDEX(LOGISTICS.ORDER_TRACK IDX_ORDER_STATUS) */
FROM LOGISTICS.ORDER_TRACK WHERE status = ‘PENDING’;

– 10. Phoenix 特有 DML
DELETE FROM LOGISTICS.ORDER_TRACK WHERE track_id = ‘TRK001’
LIMIT 1000; – Phoenix DELETE 支持 LIMIT！标准 SQL 不支持！

看完这 10 个例子，你是不是也想说：“这 TM 怎么迁？”

Phoenix SQL 迁移的"八大死局"
死局 Phoenix 特性国产库现状迁移难度
① UPSERT INTO 达梦用 MERGE INTO，金仓用 INSERT ON CONFLICT ⭐⭐⭐

② SALT_BUCKETS / COMPRESSION HBase 存储概念，关系库完全无对应 ⭐⭐⭐⭐

③ 列族语法 cf.column 关系库没有列族概念 ⭐⭐⭐⭐

④ IMMUTABLE_ROWS 关系库无对应（需要业务层保证） ⭐⭐⭐

⑤ ARRAY_ELEM / ARRAY_LENGTH 达梦/金仓数组函数名完全不同 ⭐⭐⭐

⑥ Global/Local Index 语义索引实现机制完全不同 ⭐⭐⭐⭐⭐

⑦ Phoenix Hint 语法各家国产库 Hint 语法各异 ⭐⭐⭐

⑧ DELETE … LIMIT N 标准 SQL 和多数国产库不支持 ⭐⭐⭐⭐

3200 条 SQL，每条都可能包含上述一种或多种 Phoenix 特有语法。人工逐条改写？按每人每天 50 条计算，需要 64 人天，还不算测试验证的时间。

这就是 AI 辅助 SQL 迁移引擎诞生的理由。

二、架构设计：AI + AST 变异引擎的"三层六模块"

老铁们，在动手写代码之前，必须先把架构蓝图铺开。纯靠 LLM 直接翻译 SQL 是不靠谱的——LLM 会"幻觉"，会漏掉关键语义，会对复杂嵌套查询出错。我们需要的是：

LLM 做"决策大脑"，AST 引擎做"精确手术刀"，验证框架做"质量守门员"。

🏗️ 三层架构

┌────────────────────────────────────────────────────────────────┐
│ 第三层：质量验证层 (Quality Verification Gate) │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────────┐ │
│ │ 语义等价验证 │ │ 性能回归测试 │ │ 数据一致性校验 │ │
│ │ (Semantic │ │ (Performance │ │ (Data Consistency│ │
│ │ Equivalence) │ │ Regression) │ │ Verification) │ │
│ └──────────────┘ └──────────────┘ └──────────────────┘ │
├────────────────────────────────────────────────────────────────┤
│ 第二层：AI 决策 + AST 变异层 (AI-Driven AST Mutation) │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────────┐ │
│ │ LLM 方言翻译 │ │ AST 变异规则 │ │ 迁移策略决策 │ │
│ │ Agent │ │ 引擎 │ │ 引擎 │ │
│ └──────────────┘ └──────────────┘ └──────────────────┘ │
├────────────────────────────────────────────────────────────────┤
│ 第一层：SQL 分析层 (SQL Analysis Foundation) │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────────┐ │
│ │ Phoenix SQL │ │ Phoenix 方言 │ │ SQL 特征指纹 │ │
│ │ 词法/语法分析 │ │ 特征提取 │ │ 分类器 │ │
│ └──────────────┘ └──────────────┘ └──────────────────┘ │
└────────────────────────────────────────────────────────────────┘

🧩 六大核心模块
模块职责核心技术
① Phoenix SQL 解析器词法/语法分析，构建 Phoenix 方言 AST ANTLR4 + 自定义 Phoenix Grammar

② SQL 特征指纹分类器识别 SQL 中的 Phoenix 特有语法，分类迁移难度规则引擎 + 特征向量

③ LLM 方言翻译 Agent 将 Phoenix 特有语法翻译为目标方言 OpenAI API / 本地 LLM + RAG

④ AST 变异规则引擎确定性的语法树变换（非 LLM 部分） Visitor Pattern + Rewrite Rules

⑤ 迁移策略决策引擎决定每条 SQL 的迁移路径（自动/半自动/人工）决策树 + 置信度评分

⑥ 语义等价验证框架验证迁移前后 SQL 的语义等价性 EXPLAIN 对比 + 数据集测试

三、核心模块一：Phoenix SQL 解析器（ANTLR4 方言词法/语法分析）

这是整个引擎的地基。Apache Phoenix 的 SQL 基于 Calcite，但有大量 HBase 特有的扩展语法。标准 SQL Parser（如 JSqlParser、Calcite）解析不了这些扩展。

3.1 ANTLR4 Phoenix 方言语法定义（精简核心）

// =====================================================================
// Phoenix SQL 方言词法/语法定义 (精简核心版)
// 基于 ANTLR4，覆盖 Phoenix 特有的 DDL/DML/DQL 语法
// =====================================================================

grammar PhoenixSQL;

// ─── 顶层规则 ───
phoenixStatement
: ddlStatement EOF
| dmlStatement EOF
| queryStatement EOF
| hintStatement EOF
;

// ─── DDL：Phoenix 建表（包含大量 HBase 特有选项） ───
ddlStatement
: CREATE TABLE (IF NOT EXISTS)? qualifiedName
LPAREN columnDefinition (COMMA columnDefinition)*
(COMMA tableConstraint)?
RPAREN
phoenixTableProperties? // Phoenix 特有！
| CREATE (GLOBAL | LOCAL)? INDEX indexName
ON qualifiedName LPAREN indexColumn (COMMA indexColumn)* RPAREN
(INCLUDE LPAREN columnName (COMMA columnName)* RPAREN)?
phoenixIndexProperties? // Phoenix 特有！
| CREATE SEQUENCE qualifiedName
(START WITH? INTEGER_VALUE)?
(INCREMENT BY? INTEGER_VALUE)?
(MINVALUE INTEGER_VALUE)?
(MAXVALUE INTEGER_VALUE)?
(CYCLE | NO CYCLE)?
(CACHE INTEGER_VALUE)?
| DROP TABLE (IF EXISTS)? qualifiedName (CASCADE)?
| DROP INDEX (IF EXISTS)? indexName ON qualifiedName
;

// Phoenix 建表属性（SALT_BUCKETS, IMMUTABLE_ROWS 等）
phoenixTableProperties
: tableProperty (COMMA tableProperty)*
;

tableProperty

// ─── 列定义（包含列族语法） ───
columnDefinition
: columnFamilyDotColumn dataType phoenixColumnProperties? // Phoenix 列族语法！
| columnName dataType phoenixColumnProperties?
;

// 🔑 Phoenix 独有：列族.列名的语法
// 例如：info.sender_name VARCHAR
columnFamilyDotColumn
: IDENTIFIER DOT IDENTIFIER
;

phoenixColumnProperties

NOT NULL
| NULL
| DEFAULT defaultValue
| ARRAY_BRACKET // VARCHAR[] 数组类型
;

// ─── DML：Phoenix UPSERT（核心差异点） ───
dmlStatement
: UPSERT INTO qualifiedName
(LPAREN columnName (COMMA columnName)* RPAREN)?
(VALUES LPAREN expression (COMMA expression)* RPAREN
| selectStatement)
| DELETE FROM qualifiedName
(WHERE whereClause)?
(LIMIT INTEGER_VALUE)? // Phoenix DELETE 支持 LIMIT！
;

// ─── 查询语句 ───
queryStatement
: phoenixHint? selectStatement
;

// Phoenix Hint（与 Oracle Hint 语法相似但语义不同）
phoenixHint
: HINT_START hintBody HINT_END
;

hintBody

// Phoenix 特有函数
phoenixFunction
: ARRAY_ELEM LPAREN expression COMMA INTEGER_VALUE RPAREN
| ARRAY_LENGTH LPAREN expression RPAREN
| ARRAY_APPEND LPAREN expression COMMA expression RPAREN
| ARRAY_FILL LPAREN expression COMMA INTEGER_VALUE RPAREN
| ARRAY_CAT LPAREN expression COMMA expression RPAREN
| ARRAY_TO_STRING LPAREN expression COMMA expression RPAREN
| TO_HEX LPAREN expression RPAREN
| TO_NUMBER LPAREN expression (COMMA expression)? RPAREN
| TO_DATE LPAREN expression (COMMA expression)? RPAREN
| TO_TIMESTAMP LPAREN expression (COMMA expression)? RPAREN
| TO_CHAR LPAREN expression (COMMA expression)? RPAREN
| REGEXP_SUBSTR LPAREN expression COMMA expression
(COMMA INTEGER_VALUE)? (COMMA INTEGER_VALUE)? RPAREN
| REGEXP_REPLACE LPAREN expression COMMA expression
(COMMA expression)? RPAREN
| INSTR LPAREN expression COMMA expression
(COMMA INTEGER_VALUE)? (COMMA INTEGER_VALUE)? RPAREN
| REVERSE LPAREN expression RPAREN
| ENCODE LPAREN expression COMMA expression RPAREN
| DECODE LPAREN expression (COMMA expression)+ RPAREN
| NOW LPAREN RPAREN
| CURRENT_DATE LPAREN RPAREN
| CURRENT_TIME LPAREN RPAREN
;

// ─── 词法规则（精简） ───
UPSERT : ‘UPSERT’ | ‘upsert’;
SALT_BUCKETS : ‘SALT_BUCKETS’ | ‘salt_buckets’;
IMMUTABLE_ROWS : ‘IMMUTABLE_ROWS’ | ‘immutable_rows’;
COMPRESSION : ‘COMPRESSION’ | ‘compression’;
COLUMN_ENCODED_BYTES : ‘COLUMN_ENCODED_BYTES’;
GLOBAL : ‘GLOBAL’;
LOCAL : ‘LOCAL’;
INCLUDE : ‘INCLUDE’;
ARRAY_ELEM : ‘ARRAY_ELEM’;
ARRAY_LENGTH : ‘ARRAY_LENGTH’;
ARRAY_APPEND : ‘ARRAY_APPEND’;
ARRAY_FILL : ‘ARRAY_FILL’;
ARRAY_CAT : ‘ARRAY_CAT’;
ARRAY_TO_STRING : ‘ARRAY_TO_STRING’;
TO_HEX : ‘TO_HEX’;
RANGE_SCAN : ‘RANGE_SCAN’;
SKIP_SCAN : ‘SKIP_SCAN’;
FULL_SCAN : ‘FULL_SCAN’;
HINT_START : ‘/+';
HINT_END : '/’;
ARRAY_BRACKET : ‘[]’;

// 基本 SQL 词法（省略 CREATE, TABLE, SELECT 等标准关键词）
// …

3.2 Phoenix SQL 特征指纹提取器

package com.dragonmigrate.phoenix.analysis;

import com.dragonmigrate.phoenix.parser.;
import com.dragonmigrate.phoenix.ast.;
import java.util.*;
import java.util.stream.Collectors;

/**

🟢 Phoenix SQL 特征指纹提取器 (Feature Fingerprint Extractor)

【设计思想】
不是所有 Phoenix SQL 都需要复杂迁移。大量标准 SQL（如 SELECT col FROM table WHERE x = 1）
在任何国产库上都能直接运行。
特征指纹提取器的职责是：
识别一条 SQL 中包含哪些 Phoenix 特有语法（特征标签）
为每条 SQL 计算迁移难度分数（0-100）
根据特征组合，推荐最优的迁移路径
特征标签体系（共 47 个标签，分 5 大类）：
┌─────────────┬─────────────────────────────────────────────────┐
│ 分类 │ 特征标签示例 │
├─────────────┼─────────────────────────────────────────────────┤
│ DDL-HBase │ SALT_BUCKETS, IMMUTABLE_ROWS, COMPRESSION, │
│ │ COLUMN_FAMILY, DISABLE_WAL, MULTI_TENANT │
├─────────────┼─────────────────────────────────────────────────┤
│ DML-Phoenix │ UPSERT_INTO, UPSERT_SELECT, DELETE_LIMIT │
├─────────────┼─────────────────────────────────────────────────┤
│ 索引 │ GLOBAL_INDEX, LOCAL_INDEX, COVERED_INDEX, │
│ │ INDEX_INCLUDE, FUNCTIONAL_INDEX │
├─────────────┼─────────────────────────────────────────────────┤
│ 函数 │ ARRAY_FUNC, TO_HEX, REGEXP_FUNC, ENCODE_FUNC, │
│ │ SEQUENCE_NEXT_VALUE, PHOENIX_DATE_FUNC │
├─────────────┼─────────────────────────────────────────────────┤
│ Hint/优化 │ PHOENIX_HINT, RANGE_SCAN_HINT, SKIP_SCAN_HINT, │
│ │ INDEX_HINT, CACHE_HINT │
└─────────────┴─────────────────────────────────────────────────┘

@author 墨夶
@since 2026-07-01
*/
public class PhoenixFeatureExtractor {

/**
特征标签枚举（47个，按类别组织）
*/
public enum FeatureTag {
// ─── DDL-HBase 存储特性（权重高，因为完全无对应） ───
SALT_BUCKETS(“DDL”, 15, “HBase Salting 分桶策略，关系库无直接对应”),
IMMUTABLE_ROWS(“DDL”, 10, “不可变行标记，需要业务层保证或迁移文档说明”),
COMPRESSION(“DDL”, 5, “HBase 压缩算法，关系库用自身压缩机制替代”),
COLUMN_FAMILY(“DDL”, 20, “列族声明，需要拆分为多个列或忽略”),
COLUMN_ENCODED_BYTES(“DDL”, 5, “列编码字节数，关系库无对应”),
DISABLE_WAL(“DDL”, 8, “禁用 WAL，关系库无对应”),
MULTI_TENANT(“DDL”, 12, “多租户视图策略，需要重新设计”),
DEFAULT_COLUMN_FAMILY(“DDL”, 5, “默认列族声明”),
SPLIT_POINT(“DDL”, 8, “Region 预分裂点，关系库用分区表替代”),

 // ─── DML-Phoenix 写操作（权重高，语义差异大） ───
 UPSERT_INTO("DML", 18, "UPSERT 语义，不同国产库用不同语法实现"),
 UPSERT_SELECT("DML", 20, "UPSERT...SELECT 批量搬运"),
 DELETE_LIMIT("DML", 15, "DELETE...LIMIT 语法，标准 SQL 不支持"),
 MULTI_PUT("DML", 10, "批量 PUT 操作"),

 // ─── 索引（权重最高，实现差异最大） ───
 GLOBAL_INDEX("INDEX", 25, "全局索引，底层是独立 HBase 表"),
 LOCAL_INDEX("INDEX", 20, "本地索引，与数据共处一个 Region"),
 COVERED_INDEX("INDEX", 15, "覆盖索引 INCLUDE 子句"),
 INDEX_INCLUDE("INDEX", 10, "索引附加列"),
 FUNCTIONAL_INDEX("INDEX", 18, "函数索引"),
 DESC_INDEX("INDEX", 5, "降序索引"),

 // ─── 函数（权重中等，多数有等价替换） ───
 ARRAY_ELEM_FUNC("FUNC", 12, "ARRAY_ELEM 数组元素访问"),
 ARRAY_LENGTH_FUNC("FUNC", 8, "ARRAY_LENGTH 数组长度"),
 ARRAY_APPEND_FUNC("FUNC", 10, "ARRAY_APPEND 数组追加"),
 ARRAY_FILL_FUNC("FUNC", 8, "ARRAY_FILL 数组填充"),
 ARRAY_CAT_FUNC("FUNC", 10, "ARRAY_CAT 数组拼接"),
 ARRAY_TO_STRING_FUNC("FUNC", 8, "ARRAY_TO_STRING 数组转字符串"),
 TO_HEX_FUNC("FUNC", 6, "TO_HEX 十六进制转换"),
 REGEXP_SUBSTR_FUNC("FUNC", 10, "REGEXP_SUBSTR 正则子串"),
 REGEXP_REPLACE_FUNC("FUNC", 10, "REGEXP_REPLACE 正则替换"),
 ENCODE_FUNC("FUNC", 12, "ENCODE/DECODE 编码函数"),
 SEQUENCE_NEXT_VALUE("FUNC", 8, "NEXT VALUE FOR 序列语法"),
 PHOENIX_DATE_FUNC("FUNC", 6, "Phoenix 日期函数"),

 // ─── Hint/优化（权重中等，各家语法不同） ───
 PHOENIX_INDEX_HINT("HINT", 12, "Phoenix INDEX Hint"),
 RANGE_SCAN_HINT("HINT", 8, "RANGE_SCAN Hint"),
 SKIP_SCAN_HINT("HINT", 8, "SKIP_SCAN Hint"),
 FULL_SCAN_HINT("HINT", 5, "FULL_SCAN Hint"),
 NO_INDEX_HINT("HINT", 5, "NO_INDEX Hint"),
 SMALL_HINT("HINT", 3, "SMALL Hint"),
 CACHE_HINT("HINT", 5, "CACHE Hint"),
 SCAN_BATCH_HINT("HINT", 5, "SCAN_BATCH Hint"),

 // ─── 数据类型（权重低-中） ───
 ARRAY_TYPE("TYPE", 15, "ARRAY 数据类型列"),
 UNSIGNED_INT("TYPE", 5, "UNSIGNED_INT 类型"),
 UNSIGNED_LONG("TYPE", 5, "UNSIGNED_LONG 类型"),
 UNSIGNED_FLOAT("TYPE", 5, "UNSIGNED_FLOAT 类型"),
 UNSIGNED_DOUBLE("TYPE", 5, "UNSIGNED_DOUBLE 类型"),
 UNSIGNED_TINYINT("TYPE", 5, "UNSIGNED_TINYINT 类型"),
 UNSIGNED_SMALLINT("TYPE", 5, "UNSIGNED_SMALLINT 类型"),
 UNSIGNED_TIME("TYPE", 5, "UNSIGNED_TIME 类型"),
 UNSIGNED_DATE("TYPE", 5, "UNSIGNED_DATE 类型"),
 UNSIGNED_TIMESTAMP("TYPE", 5, "UNSIGNED_TIMESTAMP 类型"),
 VARBINARY_TYPE("TYPE", 8, "VARBINARY 类型"),
 ROW_TIMESTAMP("TYPE", 10, "ROW_TIMESTAMP 标记");

 private final String category;
 private final int weight;     // 迁移难度权重（0-30）
 private final String description;

 FeatureTag(String category, int weight, String description) {
     this.category = category;
     this.weight = weight;
     this.description = description;
 }

 public String getCategory() { return category; }
 public int getWeight() { return weight; }
 public String getDescription() { return description; }

}

/**
SQL 特征指纹
/
public static class SqlFingerprint {
/* 原始 SQL */
private final String rawSql;

 /** SQL 类型：DDL / DML / DQL */
 private final String sqlType;
 
 /** 检测到的 Phoenix 特有特征标签 */
 private final Set<FeatureTag> detectedTags;
 
 /** 每个特征的详细位置（行号、列号）—— 用于精准定位和修改 */
 private final Map<FeatureTag, List<SourceLocation>> tagLocations;
 
 /** 迁移难度分数（0-100，越高越难） */
 private final int difficultyScore;
 
 /** 推荐的迁移路径 */
 private final MigrationPath recommendedPath;
 
 /** 是否为纯标准 SQL（不需要任何迁移） */
 private final boolean isStandardSql;

 // 构造器、Getter 省略

}

/**
迁移路径枚举
/
public enum MigrationPath {
/* 无需迁移，直接通过 */
PASSTHROUGH(“直接通过”, “SQL 为标准语法，无需任何修改”),

 /** AST 规则引擎自动迁移（确定性变换，100% 正确） */
 AUTO_AST("自动-AST变换", "使用确定性 AST 变换规则，无需 LLM 介入"),
 
 /** LLM 辅助迁移（需要 AI 理解语义后生成等价 SQL） */
 AI_ASSISTED("AI辅助", "需要 LLM 理解 Phoenix 语义并生成目标方言 SQL"),
 
 /** 半自动迁移（AI 生成初稿 + 人工审核） */
 SEMI_AUTO("半自动", "AI 生成迁移方案，需人工确认和微调"),
 
 /** 必须人工迁移（过于复杂或无等价替换） */
 MANUAL("人工迁移", "无直接等价替换，需架构师重新设计");

 private final String label;
 private final String description;

 MigrationPath(String label, String description) {
     this.label = label;
     this.description = description;
 }

}

/**
【核心方法】提取 SQL 的 Phoenix 特征指纹

处理流程：
ANTLR4 词法分析 → Token 流
ANTLR4 语法分析 → Parse Tree
Parse Tree 遍历 → 特征标签检测
难度评分 → 迁移路径推荐
@param rawSql 原始 SQL 文本
@return SQL 特征指纹
⚠️ 性能要求：

单条 SQL 提取时间 < 30ms
3200 条 SQL 批量提取 < 60s

使用线程池并行处理
*/
public SqlFingerprint extract(String rawSql) {
Objects.requireNonNull(rawSql, “SQL 不能为 null”);

Set tags = EnumSet.noneOf(FeatureTag.class);
Map<FeatureTag, List> tagLocations = new EnumMap<>(FeatureTag.class);

// ─── Step 1: 词法分析 ───
PhoenixSQLLexer lexer = new PhoenixSQLLexer(
org.antlr.v4.runtime.CharStreams.fromString(rawSql));
org.antlr.v4.runtime.CommonTokenStream tokens =
new org.antlr.v4.runtime.CommonTokenStream(lexer);
tokens.fill();

// ─── Step 2: 快速预扫描（基于 Token 类型，O(n) 复杂度） ───
// 在进入语法分析之前，先通过 Token 类型做快速特征检测
// 这样即使语法分析失败（SQL 语法不标准），也能提取部分特征
for (org.antlr.v4.runtime.Token token : tokens.getTokens()) {
int tokenType = token.getType();

 switch (tokenType) {
     case PhoenixSQLLexer.UPSERT:
         addTag(tags, tagLocations, FeatureTag.UPSERT_INTO,
                 new SourceLocation(token.getLine(), token.getCharPositionInLine()));
         break;
     case PhoenixSQLLexer.SALT_BUCKETS:
         addTag(tags, tagLocations, FeatureTag.SALT_BUCKETS,
                 new SourceLocation(token.getLine(), token.getCharPositionInLine()));
         break;
     case PhoenixSQLLexer.IMMUTABLE_ROWS:
         addTag(tags, tagLocations, FeatureTag.IMMUTABLE_ROWS,
                 new SourceLocation(token.getLine(), token.getCharPositionInLine()));
         break;
     case PhoenixSQLLexer.COMPRESSION:
         addTag(tags, tagLocations, FeatureTag.COMPRESSION,
                 new SourceLocation(token.getLine(), token.getCharPositionInLine()));
         break;
     case PhoenixSQLLexer.GLOBAL:
         // GLOBAL 后面跟着 INDEX 才是 Global Index
         // 需要前瞻一个 Token
         int nextIdx = tokens.indexOf(token) + 1;
         if (nextIdx < tokens.size() && 
             tokens.get(nextIdx).getType() == PhoenixSQLLexer.INDEX) {
             addTag(tags, tagLocations, FeatureTag.GLOBAL_INDEX,
                     new SourceLocation(token.getLine(), token.getCharPositionInLine()));
         }
         break;
     case PhoenixSQLLexer.LOCAL:
         int nextIdxL = tokens.indexOf(token) + 1;
         if (nextIdxL < tokens.size() && 
             tokens.get(nextIdxL).getType() == PhoenixSQLLexer.INDEX) {
             addTag(tags, tagLocations, FeatureTag.LOCAL_INDEX,
                     new SourceLocation(token.getLine(), token.getCharPositionInLine()));
         }
         break;
     case PhoenixSQLLexer.ARRAY_ELEM:
         addTag(tags, tagLocations, FeatureTag.ARRAY_ELEM_FUNC,
                 new SourceLocation(token.getLine(), token.getCharPositionInLine()));
         break;
     case PhoenixSQLLexer.ARRAY_LENGTH:
         addTag(tags, tagLocations, FeatureTag.ARRAY_LENGTH_FUNC,
                 new SourceLocation(token.getLine(), token.getCharPositionInLine()));
         break;
     case PhoenixSQLLexer.TO_HEX:
         addTag(tags, tagLocations, FeatureTag.TO_HEX_FUNC,
                 new SourceLocation(token.getLine(), token.getCharPositionInLine()));
         break;
     case PhoenixSQLLexer.HINT_START:
         // Hint 需要解析内部内容
         detectHintFeatures(tokens, tokens.indexOf(token), tags, tagLocations);
         break;
     // ... 更多 Token 类型检测
 }

}

// ─── Step 3: 语法分析（构建 Parse Tree） ───
// 使用 BailErrorStrategy 避免语法错误时的交互模式
PhoenixSQLParser parser = new PhoenixSQLParser(tokens);
parser.removeErrorListeners();
parser.setErrorHandler(new org.antlr.v4.runtime.BailErrorStrategy());

PhoenixSQLParser.PhoenixStatementContext tree = null;
String sqlType = “UNKNOWN”;
try {
tree = parser.phoenixStatement();
// 判断 SQL 类型
if (tree.ddlStatement() != null) sqlType = “DDL”;
else if (tree.dmlStatement() != null) sqlType = “DML”;
else if (tree.queryStatement() != null) sqlType = “DQL”;
} catch (Exception e) {
// ⚠️ 语法分析失败：可能是 SQL 包含 ANTLR 语法未覆盖的 Phoenix 扩展
// 降级：仅使用 Step 2 的预扫描结果
sqlType = “PARSE_FAILED”;
}

// ─── Step 4: Parse Tree 深度遍历（补充预扫描遗漏的特征） ───
if (tree != null) {
PhoenixFeatureVisitor visitor = new PhoenixFeatureVisitor(tags, tagLocations);
visitor.visit(tree);
}

// ─── Step 5: 基于正则的补充检测（兜底策略） ───
// 某些特征可能因为 ANTLR 语法不完整而遗漏，用正则兜底
detectByRegex(rawSql, tags, tagLocations);

// ─── Step 6: 难度评分 ───
int difficultyScore = calculateDifficulty(tags);

// ─── Step 7: 迁移路径推荐 ───
MigrationPath path = recommendMigrationPath(tags, difficultyScore);

return new SqlFingerprint(
rawSql, sqlType, tags, tagLocations,
difficultyScore, path, tags.isEmpty()
);
}

/**
【难度评分算法】

评分公式：
baseScore = sum(tag.weight) (所有特征的权重之和)
comboBonus = 特征组合加成（某些特征组合在一起会显著增加难度）
finalScore = min(100, baseScore + comboBonus)
组合加成示例：

SALT_BUCKETS + COLUMN_FAMILY → +10（列族+分桶需要整体重新设计表结构）
GLOBAL_INDEX + COVERED_INDEX → +8（覆盖索引+全局索引的复合特性）
UPSERT_SELECT + DELETE_LIMIT → +5（DML 语义差异叠加）
*/
private int calculateDifficulty(Set tags) {
int baseScore = tags.stream().mapToInt(FeatureTag::getWeight).sum();

int comboBonus = 0;
// 组合加成检测
if (tags.contains(FeatureTag.SALT_BUCKETS) &&
tags.contains(FeatureTag.COLUMN_FAMILY)) {
comboBonus += 10;
}
if (tags.contains(FeatureTag.GLOBAL_INDEX) &&
tags.contains(FeatureTag.COVERED_INDEX)) {
comboBonus += 8;
}
if (tags.contains(FeatureTag.UPSERT_SELECT) &&
tags.contains(FeatureTag.DELETE_LIMIT)) {
comboBonus += 5;
}
if (tags.contains(FeatureTag.ARRAY_TYPE) &&
tags.contains(FeatureTag.ARRAY_ELEM_FUNC)) {
comboBonus += 7;
}

return Math.min(100, baseScore + comboBonus);
}

/**
【迁移路径推荐】

决策逻辑：

难度 0（无特征）→ PASSTHROUGH
难度 1-25 且仅包含确定性变换特征 → AUTO_AST
难度 26-60 → AI_ASSISTED
难度 61-80 → SEMI_AUTO
难度 81-100 → MANUAL

⚠️ 特殊规则：

包含 SALT_BUCKETS 或 MULTI_TENANT 的 SQL，即使难度不高也推荐 SEMI_AUTO
（因为这些特性的迁移方案需要架构师确认）
包含 COLUMN_FAMILY 的 DDL，必须推荐 SEMI_AUTO 或 MANUAL
*/
private MigrationPath recommendMigrationPath(Set tags, int difficulty) {
// 特殊规则优先
if (tags.contains(FeatureTag.SALT_BUCKETS) ||
tags.contains(FeatureTag.MULTI_TENANT)) {
return difficulty > 60 ? MigrationPath.MANUAL : MigrationPath.SEMI_AUTO;
}
if (tags.contains(FeatureTag.COLUMN_FAMILY)) {
return difficulty > 40 ? MigrationPath.MANUAL : MigrationPath.SEMI_AUTO;
}

// 通用规则
if (difficulty == 0) return MigrationPath.PASSTHROUGH;
if (difficulty <= 25 && isAllDeterministic(tags)) return MigrationPath.AUTO_AST;
if (difficulty <= 60) return MigrationPath.AI_ASSISTED;
if (difficulty <= 80) return MigrationPath.SEMI_AUTO;
return MigrationPath.MANUAL;
}

/**
判断所有特征是否都是"确定性变换"的
（即可以用 AST 规则引擎 100% 正确转换，不需要 LLM 理解语义）
*/
private boolean isAllDeterministic(Set tags) {
Set nonDeterministic = EnumSet.of(
FeatureTag.SALT_BUCKETS, FeatureTag.MULTI_TENANT,
FeatureTag.COLUMN_FAMILY, FeatureTag.IMMUTABLE_ROWS,
FeatureTag.GLOBAL_INDEX, FeatureTag.LOCAL_INDEX
);
return tags.stream().noneMatch(nonDeterministic::contains);
}

private void addTag(Set tags,
Map<FeatureTag, List> locations,
FeatureTag tag, SourceLocation loc) {
tags.add(tag);
locations.computeIfAbsent(tag, k -> new ArrayList<>()).add(loc);
}

private void detectHintFeatures(
org.antlr.v4.runtime.CommonTokenStream tokens, int hintStartIdx,
Set tags, Map<FeatureTag, List> locations) {
// 从 HINT_START 向后扫描，直到 HINT_END
for (int i = hintStartIdx + 1; i < tokens.size(); i++) {
org.antlr.v4.runtime.Token t = tokens.get(i);
if (t.getType() == PhoenixSQLLexer.HINT_END) break;

     String text = t.getText().toUpperCase();
     switch (text) {
         case "INDEX" -> addTag(tags, locations, FeatureTag.PHOENIX_INDEX_HINT,
                 new SourceLocation(t.getLine(), t.getCharPositionInLine()));
         case "RANGE_SCAN" -> addTag(tags, locations, FeatureTag.RANGE_SCAN_HINT,
                 new SourceLocation(t.getLine(), t.getCharPositionInLine()));
         case "SKIP_SCAN" -> addTag(tags, locations, FeatureTag.SKIP_SCAN_HINT,
                 new SourceLocation(t.getLine(), t.getCharPositionInLine()));
         case "FULL_SCAN" -> addTag(tags, locations, FeatureTag.FULL_SCAN_HINT,
                 new SourceLocation(t.getLine(), t.getCharPositionInLine()));
         case "NO_INDEX" -> addTag(tags, locations, FeatureTag.NO_INDEX_HINT,
                 new SourceLocation(t.getLine(), t.getCharPositionInLine()));
         case "SMALL" -> addTag(tags, locations, FeatureTag.SMALL_HINT,
                 new SourceLocation(t.getLine(), t.getCharPositionInLine()));
         case "CACHE" -> addTag(tags, locations, FeatureTag.CACHE_HINT,
                 new SourceLocation(t.getLine(), t.getCharPositionInLine()));
         case "SCAN_BATCH" -> addTag(tags, locations, FeatureTag.SCAN_BATCH_HINT,
                 new SourceLocation(t.getLine(), t.getCharPositionInLine()));
     }
 }

}

/**
【正则兜底检测】
对于 ANTLR 语法未覆盖的 Phoenix 扩展，使用正则表达式做补充检测
/
private void detectByRegex(String sql, Set tags,
Map<FeatureTag, List> locations) {
// 检测 UNSIGNED 类型
if (sql.matches("(?i).\bUNSIGNED_(INT|LONG|FLOAT|DOUBLE|TINYINT|SMALLINT|" +
“TIME|DATE|TIMESTAMP)\b.*”)) {
// 细化检测具体类型
if (sql.matches(“(?i)\bUNSIGNED_INT\b.”))
addTag(tags, locations, FeatureTag.UNSIGNED_INT, SourceLocation.UNKNOWN);
if (sql.matches(“(?i)\bUNSIGNED_LONG\b.”))
addTag(tags, locations, FeatureTag.UNSIGNED_LONG, SourceLocation.UNKNOWN);
// … 其他 UNSIGNED 类型
}

 // 检测 VARBINARY
 if (sql.matches("(?i)\bVARBINARY\b.")) {
     addTag(tags, locations, FeatureTag.VARBINARY_TYPE, SourceLocation.UNKNOWN);
 }

 // 检测 ROW_TIMESTAMP
 if (sql.matches("(?i)\bROW_TIMESTAMP\b.")) {
     addTag(tags, locations, FeatureTag.ROW_TIMESTAMP, SourceLocation.UNKNOWN);
 }

 // 检测列族语法：identifier.identifier 在列定义上下文中
 // ⚠️ 注意：这里的正则可能误判（如 schema.table），需要结合语法分析结果过滤
 if (sql.matches("(?i).*\w+\.\w+\s+(VARCHAR|INTEGER|DOUBLE|BIGINT|BOOLEAN|" +
         "TIMESTAMP|DATE|DECIMAL|FLOAT|CHAR|BINARY).*")) {
     // 仅在 DDL 上下文中才标记为列族
     if (sql.toUpperCase().contains("CREATE TABLE")) {
         addTag(tags, locations, FeatureTag.COLUMN_FAMILY, SourceLocation.UNKNOWN);
     }
 }

}
}

四、核心模块二：AST 变异规则引擎（确定性 SQL 变换）

这是引擎的手术刀。对于 Phoenix 中那些有明确等价替换的语法，我们用确定性的 AST 变换规则来转换，不需要 LLM 介入——确保 100% 正确。

4.1 变异规则注册表

package com.dragonmigrate.phoenix.mutation;

import com.dragonmigrate.core.TargetDialect;
import com.dragonmigrate.phoenix.ast.;
import java.util.;

/**

🟡 AST 变异规则引擎 (AST Mutation Rule Engine)

【设计思想】
Phoenix SQL 中的特性可以分为两类：
1️⃣ 确定性变换（Deterministic Transform）：
有明确的、唯一的等价替换，不需要理解业务语义。
例如：

UPSERT INTO → MERGE INTO (达梦) / INSERT ON CONFLICT (金仓)
TO_HEX(x) → HEX(x) (MySQL) / RAWTOHEX(x) (达梦)
ARRAY_ELEM(arr, 1) → arr[1] (金仓) / JSON_EXTRACT (达梦)
DELETE…LIMIT N → DELETE…WHERE ROWNUM <= N (达梦)

2️⃣ 语义依赖变换（Semantic-Dependent Transform）：
需要理解业务语义才能确定替换方案。
例如：

SALT_BUCKETS=16 → 可能是分区表、可能是 Hash 索引，取决于业务场景
GLOBAL INDEX → 可能是普通索引、可能是物化视图，取决于查询模式
IMMUTABLE_ROWS → 可能需要应用层保证，可能需要触发器

AST 变异规则引擎只处理第一类。第二类交给 LLM Agent。
【架构模式】
采用策略链(Chain of Strategy) 模式：

每个规则是一个独立的 Strategy
规则按优先级排序执行
每个规则只负责一种 Phoenix 特征的变换
规则之间互不干扰（正交性）

@author 墨夶
@since 2026-07-01
*/
public class AstMutationEngine {

/** 目标方言 */
private final TargetDialect targetDialect;

/** 有序规则链 */
private final List ruleChain;

/** 变异统计（用于生成迁移报告） */
private final MutationStatistics statistics;

public AstMutationEngine(TargetDialect targetDialect) {
this.targetDialect = Objects.requireNonNull(targetDialect);
this.statistics = new MutationStatistics();
this.ruleChain = buildRuleChain(targetDialect);
}

/**
【构建规则链】
按执行优先级排序：DDL 规则 → DML 规则 → 函数规则 → Hint 规则
- ⚠️ 顺序很重要！
  例如：必须先处理 UPSERT → INSERT/MERGE，再处理函数替换
  因为 UPSERT 内部的函数引用需要在正确的语句上下文中替换
  */
  private List buildRuleChain(TargetDialect dialect) {
  List rules = new ArrayList<>();
  
  // ─── DDL 规则 ───
  rules.add(new StripHBasePropertiesRule()); // 剥离 HBase 存储属性
  rules.add(new ColumnFamilyFlattenRule()); // 列族展平
  rules.add(new UnsignedTypeMappingRule()); // UNSIGNED 类型映射
  rules.add(new ArrayTypeMappingRule()); // ARRAY 类型映射
  rules.add(new SequenceTransformRule(dialect)); // 序列语法转换
  
  // ─── DML 规则 ───
  rules.add(new UpsertTransformRule(dialect)); // UPSERT 转换
  rules.add(new DeleteLimitTransformRule(dialect)); // DELETE…LIMIT 转换
  
  // ─── 函数规则 ───
  rules.add(new ArrayFunctionTransformRule(dialect)); // ARRAY 函数
  rules.add(new HexFunctionTransformRule(dialect)); // TO_HEX 函数
  rules.add(new RegexpFunctionTransformRule(dialect)); // REGEXP 函数
  rules.add(new EncodeFunctionTransformRule(dialect)); // ENCODE/DECODE 函数
  rules.add(new DateFunctionTransformRule(dialect)); // 日期函数
  
  // ─── Index 规则 ───
  rules.add(new IndexTransformRule(dialect)); // 索引语法转换
  
  // ─── Hint 规则 ───
  rules.add(new HintTransformRule(dialect)); // Hint 转换或移除
  
  return rules;
  }
/**
【核心方法】对 AST 执行全部变异规则
- @param ast 原始 Phoenix SQL 的 AST
  @param fingerprint SQL 特征指纹（用于跳过不相关的规则，提升性能）
  @return 变异后的 AST + 变异报告
- ⚠️ 幂等性保证：
  同一条 SQL 多次执行变异规则，结果必须一致。
  每条规则在修改 AST 之前，必须检查目标节点是否已经被其他规则修改过。
- 🚀 性能目标：
- 单条 SQL 变异 < 20ms
- 全部规则遍历 < 50ms（即使大部分规则不触发）
  */
  public MutationResult mutate(PhoenixAstNode ast, SqlFingerprint fingerprint) {
  Objects.requireNonNull(ast, “AST 不能为 null”);
  
  MutationContext context = new MutationContext(targetDialect, fingerprint, statistics);
  List appliedMutations = new ArrayList<>();
  
  for (MutationRule rule : ruleChain) {
  // ⚠️ 性能优化：如果 fingerprint 中没有此规则关心的特征标签，跳过
  if (!rule.shouldApply(fingerprint)) {
  continue;
  }
```
 try {
     // 深拷贝 AST 节点，防止规则失败时破坏原始 AST
     PhoenixAstNode astSnapshot = ast.deepCopy();

     boolean applied = rule.apply(ast, context);
     if (applied) {
         appliedMutations.add(new MutationRecord(
                 rule.getName(),
                 rule.getDescription(),
                 astSnapshot.toSql(), // 变异前 SQL
                 ast.toSql()          // 变异后 SQL
         ));
     }
 } catch (MutationException e) {
     // ⚠️ 规则执行失败：记录错误，但不中断后续规则
     // 失败的规则对应的特征会留给 LLM Agent 处理
     context.addError(rule.getName(), e.getMessage());
     statistics.incrementFailure(rule.getName());
 }
```
  }
  
  return new MutationResult(
  ast,
  ast.toSql(),
  appliedMutations,
  context.getErrors(),
  context.getUnresolvedTags() // 未解决的特征标签（留给 LLM）
  );
  }
  }

/**

变异规则接口 (Mutation Rule Interface)

/
interface MutationRule {
/* 规则名称 */
String getName();

/** 规则描述 */
String getDescription();

/** 根据特征指纹判断此规则是否需要执行 */
boolean shouldApply(SqlFingerprint fingerprint);

/** 执行变异，返回是否成功应用 */
boolean apply(PhoenixAstNode ast, MutationContext context) throws MutationException;

}

4.2 UPSERT 变异规则（核心实现 ⭐⭐⭐⭐⭐）

package com.dragonmigrate.phoenix.mutation.rules;

import com.dragonmigrate.core.TargetDialect;
import com.dragonmigrate.phoenix.ast.;
import com.dragonmigrate.phoenix.mutation.;
import com.dragonmigrate.phoenix.analysis.PhoenixFeatureExtractor.FeatureTag;

/**

🔴 UPSERT 变异规则 (UPSERT Transform Rule)

【Phoenix UPSERT 语义】
UPSERT INTO table (col1, col2) VALUES (val1, val2)
语义：如果 RowKey 存在则 UPDATE，不存在则 INSERT
这与标准 SQL 的 INSERT 不同（INSERT 在 PK 冲突时会报错）
【各目标方言的等价实现】
┌──────────────┬─────────────────────────────────────────────────────────────┐
│ 目标方言 │ 等价 SQL │
├──────────────┼─────────────────────────────────────────────────────────────┤
│ 达梦 DM8 │ MERGE INTO table t USING (SELECT val1 AS col1, …) s │
│ │ ON (t.pk = s.pk) │
│ │ WHEN MATCHED THEN UPDATE SET t.col1 = s.col1 │
│ │ WHEN NOT MATCHED THEN INSERT (col1) VALUES (s.col1) │
├──────────────┼─────────────────────────────────────────────────────────────┤
│ 人大金仓 │ INSERT INTO table (col1, col2) VALUES (val1, val2) │
│ KingbaseES │ ON CONFLICT (pk_columns) │
│ │ DO UPDATE SET col1 = EXCLUDED.col1, col2 = EXCLUDED.col2 │
├──────────────┼─────────────────────────────────────────────────────────────┤
│ OceanBase │ INSERT INTO table (col1, col2) VALUES (val1, val2) │
│ (MySQL模式) │ ON DUPLICATE KEY UPDATE col1 = VALUES(col1), col2 = VALUES(col2) │
├──────────────┼─────────────────────────────────────────────────────────────┤
│ TiDB │ INSERT INTO table (col1, col2) VALUES (val1, val2) │
│ │ ON DUPLICATE KEY UPDATE col1 = VALUES(col1) │
└──────────────┴─────────────────────────────────────────────────────────────┘
⚠️ 关键差异点：
Phoenix UPSERT 的"冲突判断"基于 RowKey（即 PRIMARY KEY）
所有方言的等价实现也都基于 PK 判断，语义一致
1. Phoenix UPSERT 不需要显式指定冲突列（隐式使用 PK）
  但金仓的 ON CONFLICT 需要显式指定冲突列
  → 需要查询元数据获取 PK 列名
1. Phoenix UPSERT…SELECT 批量搬运的情况更复杂
  → 需要生成批量 MERGE/INSERT 语句
@author 墨夶
@since 2026-07-01
*/
public class UpsertTransformRule implements MutationRule {

private final TargetDialect targetDialect;

public UpsertTransformRule(TargetDialect targetDialect) {
this.targetDialect = targetDialect;
}

@Override
public String getName() { return “UpsertTransformRule”; }

@Override
public String getDescription() {
return “将 Phoenix UPSERT INTO 转换为目标方言的等价 upsert 语法”;
}

@Override
public boolean shouldApply(SqlFingerprint fingerprint) {
// 仅在检测到 UPSERT_INTO 或 UPSERT_SELECT 特征时触发
return fingerprint.getDetectedTags().contains(FeatureTag.UPSERT_INTO)
|| fingerprint.getDetectedTags().contains(FeatureTag.UPSERT_SELECT);
}

@Override
public boolean apply(PhoenixAstNode ast, MutationContext context) throws MutationException {
// 查找 AST 中的 UPSERT 节点
List upsertNodes = ast.findNodesByType(AstNodeType.UPSERT_STATEMENT);
if (upsertNodes.isEmpty()) return false;
```
 boolean anyApplied = false;

 for (AstNode upsertNode : upsertNodes) {
     // 提取 UPSERT 的结构信息
     String tableName = upsertNode.getChild("targetTable").getText();
     List<String> columns = upsertNode.getChild("columnList")
             .getChildren().stream()
             .map(AstNode::getText)
             .collect(java.util.stream.Collectors.toList());

     // 判断是 UPSERT...VALUES 还是 UPSERT...SELECT
     AstNode valuesNode = upsertNode.getChild("valuesClause");
     AstNode selectNode = upsertNode.getChild("selectClause");

     String transformedSql;
     if (valuesNode != null) {
         // UPSERT INTO table (cols) VALUES (vals)
         List<String> values = valuesNode.getChildren().stream()
                 .map(AstNode::getText)
                 .collect(java.util.stream.Collectors.toList());
         transformedSql = transformUpsertValues(tableName, columns, values, context);
     } else if (selectNode != null) {
         // UPSERT INTO table SELECT ...
         String selectSql = selectNode.toSql();
         transformedSql = transformUpsertSelect(tableName, columns, selectSql, context);
     } else {
         throw new MutationException("UPSERT 语句缺少 VALUES 或 SELECT 子句");
     }

     // 替换 AST 节点
     upsertNode.replaceWith(new RawSqlNode(transformedSql));
     anyApplied = true;
 }

 return anyApplied;
```
}

/**
【UPSERT…VALUES 变换】
- 生成目标方言的等价 SQL
- @param tableName 目标表名
  @param columns 列名列表
  @param values 值列表
  @param context 变异上下文（包含元数据访问器）
  @return 目标方言 SQL
- ⚠️ 易错点 1：
  Phoenix UPSERT 如果省略了列列表，表示使用所有列。
  此时需要从元数据中获取完整的列列表。
- ⚠️ 易错点 2：
  达梦的 MERGE INTO 要求 USING 子句是一个查询（不能直接写值）。
  所以 VALUES (v1, v2) 需要改写为 SELECT v1 AS c1, v2 AS c2 FROM DUAL。
- ⚠️ 易错点 3：
  金仓的 ON CONFLICT 需要指定冲突列（通常是 PK）。
  如果 Phoenix 表没有显式 PK（使用 RowKey），需要从元数据推断。
- ⚠️ 边界条件：
- columns 为空 → 全列 UPSERT，需要查元数据
- values 中包含子查询 → 需要特殊处理
- tableName 包含 schema 前缀 → 需要保留
  */
  private String transformUpsertValues(String tableName, List columns,
  List values,
  MutationContext context) throws MutationException {
  // 如果列列表为空，从元数据获取
  if (columns.isEmpty()) {
  columns = context.getMetadataProvider().getAllColumns(tableName);
  if (columns.isEmpty()) {
  throw new MutationException(“无法获取表 " + tableName + " 的列信息”);
  }
  }
  
  // 获取主键列（用于冲突判断）
  List pkColumns = context.getMetadataProvider().getPrimaryKeyColumns(tableName);
  if (pkColumns.isEmpty()) {
  // ⚠️ Phoenix 表必须有 PK（RowKey），如果没有查到，可能元数据不完整
  // 降级策略：使用第一个列作为冲突判断列
  pkColumns = List.of(columns.get(0));
  context.addWarning(“表 " + tableName + " 未找到主键，使用第一列作为冲突判断列”);
  }
  
  // 非 PK 列（用于 UPDATE SET 子句）
  List nonPkColumns = columns.stream()
  .filter(c -> !pkColumns.contains©)
  .collect(java.util.stream.Collectors.toList());
  
  return switch (context.getTargetDialect()) {
  case DM -> generateDmMerge(tableName, columns, values, pkColumns, nonPkColumns);
  case KINGBASE -> generateKingbaseInsertOnConflict(
  tableName, columns, values, pkColumns, nonPkColumns);
  case OCEANBASE_MYSQL -> generateObMysqlInsertOnDuplicate(
  tableName, columns, values, nonPkColumns);
  case TIDB -> generateObMysqlInsertOnDuplicate(
  tableName, columns, values, nonPkColumns); // TiDB 语法与 MySQL 相同
  default -> throw new MutationException(
  "不支持的目标方言: " + context.getTargetDialect());
  };
  }
/**
【达梦 DM8 MERGE INTO 生成】
- 生成格式：
  MERGE INTO schema.table t
  USING (SELECT v1 AS col1, v2 AS col2, … FROM DUAL) s
  ON (t.pk1 = s.pk1 AND t.pk2 = s.pk2)
  WHEN MATCHED THEN
  UPDATE SET t.non_pk1 = s.non_pk1, t.non_pk2 = s.non_pk2
  WHEN NOT MATCHED THEN
  INSERT (col1, col2, …) VALUES (s.col1, s.col2, …)
- ⚠️ 易错点：
  如果所有列都是 PK 列（nonPkColumns 为空），
  MERGE 的 WHEN MATCHED THEN UPDATE 子句是空的，达梦会报错！
  解决方案：使用 WHEN MATCHED THEN UPDATE SET pk1 = s.pk1（自更新，无实际效果）
  或者改用 INSERT…WHERE NOT EXISTS 模式
- 1. USING 子句的 SELECT 必须 FROM DUAL（达梦要求）
- 1. 值中如果包含单引号，需要转义
    */
    private String generateDmMerge(String tableName, List columns,
    List values, List pkColumns,
    List nonPkColumns) {
    StringBuilder sb = new StringBuilder();
  sb.append(“MERGE INTO “).append(tableName).append(” tn”);
  sb.append("USING (SELECT ");
  
  // USING 子句：SELECT v1 AS col1, v2 AS col2, … FROM DUAL
  for (int i = 0; i < columns.size(); i++) {
  if (i > 0) sb.append(“, “);
  sb.append(values.get(i)).append(” AS “).append(columns.get(i));
  }
  sb.append(” FROM DUAL) sn”);
  
  // ON 子句：PK 匹配条件
  sb.append(“ON (”);
  for (int i = 0; i < pkColumns.size(); i++) {
  if (i > 0) sb.append(" AND “);
  sb.append(“t.”).append(pkColumns.get(i))
  .append(” = s.“).append(pkColumns.get(i));
  }
  sb.append(”)n");
  
  // WHEN MATCHED 子句
  if (!nonPkColumns.isEmpty()) {
  sb.append(“WHEN MATCHED THENn UPDATE SET “);
  for (int i = 0; i < nonPkColumns.size(); i++) {
  if (i > 0) sb.append(”, “);
  String col = nonPkColumns.get(i);
  sb.append(“t.”).append(col).append(” = s.”).append(col);
  }
  sb.append(“n”);
  } else {
  // 所有列都是 PK，无需 UPDATE
  // 达梦不允许空的 UPDATE SET，使用一个无害的自更新
  sb.append(“WHEN MATCHED THENn UPDATE SET t.”)
  .append(pkColumns.get(0)).append(" = s.").append(pkColumns.get(0)).append(“n”);
  }
  
  // WHEN NOT MATCHED 子句
  sb.append(“WHEN NOT MATCHED THENn INSERT (”);
  sb.append(String.join(“, “, columns));
  sb.append(”)n VALUES (”);
  for (int i = 0; i < columns.size(); i++) {
  if (i > 0) sb.append(“, “);
  sb.append(“s.”).append(columns.get(i));
  }
  sb.append(”)”);
  
  return sb.toString();
  }
/**
【人大金仓 KingbaseES INSERT ON CONFLICT 生成】
- 生成格式（PostgreSQL 兼容语法）：
  INSERT INTO schema.table (col1, col2, …)
  VALUES (val1, val2, …)
  ON CONFLICT (pk1, pk2) DO UPDATE SET
  non_pk1 = EXCLUDED.non_pk1,
  non_pk2 = EXCLUDED.non_pk2
- ⚠️ 易错点：
  ON CONFLICT 的冲突列必须有唯一约束（PK 天然满足）
  EXCLUDED 是 PostgreSQL 的伪表，引用"被排除的行"（即要插入的行）
  如果所有列都是 PK，DO UPDATE SET 不能引用 PK 列
  → 使用 DO NOTHING（仅当不存在时插入，存在时不做任何操作）
  这与 Phoenix UPSERT 的语义有微妙差异！
  Phoenix UPSERT 在全 PK 场景下，重复 UPSERT 是幂等的（写入相同值）
  DO NOTHING 也是幂等的 → 语义等价！✅
  */
  private String generateKingbaseInsertOnConflict(String tableName, List columns,
  List values, List pkColumns,
  List nonPkColumns) {
  StringBuilder sb = new StringBuilder();
  
  sb.append(“INSERT INTO “).append(tableName).append(” (”);
  sb.append(String.join(“, “, columns));
  sb.append(”)nVALUES (”);
  sb.append(String.join(“, “, values));
  sb.append(”)nON CONFLICT (”);
  sb.append(String.join(", “, pkColumns));
  sb.append(”) ");
  
  if (!nonPkColumns.isEmpty()) {
  sb.append(“DO UPDATE SETn “);
  for (int i = 0; i < nonPkColumns.size(); i++) {
  if (i > 0) sb.append(”,n “);
  String col = nonPkColumns.get(i);
  sb.append(col).append(” = EXCLUDED.”).append(col);
  }
  } else {
  // 全 PK 场景：DO NOTHING
  sb.append(“DO NOTHING”);
  }
  
  return sb.toString();
  }
/**
【OceanBase MySQL 模式 INSERT ON DUPLICATE KEY UPDATE 生成】
- 生成格式：
  INSERT INTO schema.table (col1, col2)
  VALUES (val1, val2)
  ON DUPLICATE KEY UPDATE
  non_pk1 = VALUES(non_pk1),
  non_pk2 = VALUES(non_pk2)
- ⚠️ 易错点：
  ON DUPLICATE KEY UPDATE 不需要显式指定冲突列（自动使用所有唯一键）
  这与 Phoenix UPSERT（基于 RowKey）语义基本等价
  VALUES() 函数引用"要插入的值"
  注意：MySQL 8.0.20+ 已弃用 VALUES() 函数，推荐使用别名语法
  但 OceanBase 目前仍然支持 VALUES() 函数
  */
  private String generateObMysqlInsertOnDuplicate(String tableName, List columns,
  List values,
  List nonPkColumns) {
  StringBuilder sb = new StringBuilder();
  
  sb.append(“INSERT INTO “).append(tableName).append(” (”);
  sb.append(String.join(“, “, columns));
  sb.append(”)nVALUES (”);
  sb.append(String.join(", “, values));
  sb.append(”)nON DUPLICATE KEY UPDATEn ");
  
  if (!nonPkColumns.isEmpty()) {
  for (int i = 0; i < nonPkColumns.size(); i++) {
  if (i > 0) sb.append(“,n “);
  String col = nonPkColumns.get(i);
  sb.append(col).append(” = VALUES(”).append(col).append(“)”);
  }
  } else {
  // 全 PK 场景：自更新（幂等操作）
  // 使用一个无害的赋值：col = col
  sb.append(columns.get(0)).append(" = ").append(columns.get(0));
  }
  
  return sb.toString();
  }
/**
【UPSERT…SELECT 变换】
将 UPSERT INTO target SELECT … FROM source 转换为批量 MERGE/INSERT
- 这比 UPSERT…VALUES 复杂得多，因为 SELECT 的结果可能有多行
- @param tableName 目标表名
  @param columns 目标列列表
  @param selectSql SELECT 子句 SQL
  @param context 变异上下文
  */
  private String transformUpsertSelect(String tableName, List columns,
  String selectSql,
  MutationContext context) throws MutationException {
  List pkColumns = context.getMetadataProvider().getPrimaryKeyColumns(tableName);
  List nonPkColumns = columns.stream()
  .filter(c -> !pkColumns.contains©)
  .collect(java.util.stream.Collectors.toList());
  
  return switch (context.getTargetDialect()) {
  case DM -> {
  // 达梦：MERGE INTO … USING (SELECT …) s ON …
  StringBuilder sb = new StringBuilder();
  sb.append(“MERGE INTO “).append(tableName).append(” tn”);
  sb.append(“USING (n “).append(selectSql).append(“n) sn”);
  sb.append(“ON (”);
  for (int i = 0; i < pkColumns.size(); i++) {
  if (i > 0) sb.append(” AND “);
  sb.append(“t.”).append(pkColumns.get(i))
  .append(” = s.”).append(pkColumns.get(i));
  }
  sb.append(“)n”);
```
     if (!nonPkColumns.isEmpty()) {
         sb.append("WHEN MATCHED THENn  UPDATE SET ");
         for (int i = 0; i < nonPkColumns.size(); i++) {
             if (i > 0) sb.append(", ");
             sb.append("t.").append(nonPkColumns.get(i))
                     .append(" = s.").append(nonPkColumns.get(i));
         }
         sb.append("n");
     }

     sb.append("WHEN NOT MATCHED THENn  INSERT (");
     sb.append(String.join(", ", columns));
     sb.append(")n  VALUES (");
     sb.append(columns.stream().map(c -> "s." + c)
             .collect(java.util.stream.Collectors.joining(", ")));
     sb.append(")");
     yield sb.toString();
 }
 case KINGBASE -> {
     // 金仓：INSERT INTO ... SELECT ... ON CONFLICT DO UPDATE
     StringBuilder sb = new StringBuilder();
     sb.append("INSERT INTO ").append(tableName).append(" (");
     sb.append(String.join(", ", columns));
     sb.append(")n").append(selectSql);
     sb.append("nON CONFLICT (");
     sb.append(String.join(", ", pkColumns));
     sb.append(") ");
     if (!nonPkColumns.isEmpty()) {
         sb.append("DO UPDATE SETn  ");
         for (int i = 0; i < nonPkColumns.size(); i++) {
             if (i > 0) sb.append(",n  ");
             sb.append(nonPkColumns.get(i))
                     .append(" = EXCLUDED.").append(nonPkColumns.get(i));
         }
     } else {
         sb.append("DO NOTHING");
     }
     yield sb.toString();
 }
 case OCEANBASE_MYSQL, TIDB -> {
     // OceanBase MySQL：INSERT INTO ... SELECT ... ON DUPLICATE KEY UPDATE
     StringBuilder sb = new StringBuilder();
     sb.append("INSERT INTO ").append(tableName).append(" (");
     sb.append(String.join(", ", columns));
     sb.append(")n").append(selectSql);
     sb.append("nON DUPLICATE KEY UPDATEn  ");
     if (!nonPkColumns.isEmpty()) {
         for (int i = 0; i < nonPkColumns.size(); i++) {
             if (i > 0) sb.append(",n  ");
             sb.append(nonPkColumns.get(i))
                     .append(" = VALUES(").append(nonPkColumns.get(i)).append(")");
         }
     } else {
         sb.append(columns.get(0)).append(" = ").append(columns.get(0));
     }
     yield sb.toString();
 }
 default -> throw new MutationException("不支持的目标方言: " + context.getTargetDialect());
```
  };
  }
  }

五、核心模块三：LLM 方言翻译 Agent（AI 决策大脑）

AST 规则引擎处理确定性变换，但 Phoenix 中有大量特性需要理解业务语义才能迁移。这就是 LLM Agent 的用武之地。

5.1 AI Agent 核心框架

package com.dragonmigrate.phoenix.ai;

import com.dragonmigrate.core.TargetDialect;
import com.dragonmigrate.phoenix.analysis.PhoenixFeatureExtractor.FeatureTag;
import com.dragonmigrate.phoenix.mutation.MutationResult;
import com.dragonmigrate.phoenix.ast.;
import java.util.;
import java.util.concurrent.*;

/**

🔴 LLM 方言翻译 Agent (AI-Powered Dialect Translation Agent)

【设计思想】
为什么不能全靠 LLM？
LLM 可能产生"幻觉"（编造不存在的函数名）
LLM 对复杂嵌套 SQL 的改写可能遗漏子查询
LLM 不知道目标数据库的精确版本和配置
LLM 无法保证"语义等价性"
为什么不能全靠 AST 规则？
Phoenix 的 SALT_BUCKETS 语义取决于业务访问模式
IMMUTABLE_ROWS 的替代方案取决于数据是否真的不可变
GLOBAL INDEX vs LOCAL INDEX 的替代取决于查询模式
所以我们的策略是：

AST 规则引擎先做确定性变换（解决 80% 的简单问题）
LLM Agent 处理 AST 规则无法解决的"语义依赖变换"（剩余 20%）
LLM 输出必须经过"验证框架"确认等价性

【Agent 架构】
采用 ReAct (Reasoning + Acting) 模式：
Reasoning：分析 Phoenix 特性的业务语义
Acting：生成目标方言 SQL
Verification：验证语义等价性
Reflection：如果验证失败，反思并重新生成
【安全护栏】

Token 限制：单次 LLM 调用 < 4000 tokens
超时控制：单次 LLM 调用 < 30s
重试限制：最多重试 3 次
输出验证：LLM 生成的 SQL 必须通过语法解析
人工兜底：3 次重试仍失败 → 标记为 MANUAL，提交人工审核

@author 墨夶
@since 2026-07-01
*/
public class LlmdialectTranslationAgent {

private final LlmClient llmClient; // LLM API 客户端
private final TargetDialect targetDialect; // 目标方言
private final KnowledgeBase knowledgeBase; // Phoenix→目标方言的知识库（RAG）
private final SemanticVerifier verifier; // 语义等价验证器
private final MetadataProvider metadataProvider; // 元数据提供者

/** 最大重试次数 */
private static final int MAX_RETRIES = 3;

/** LLM 调用超时（秒） */
private static final int LLM_TIMEOUT_SECONDS = 30;

public LlmdialectTranslationAgent(LlmClient llmClient,
TargetDialect targetDialect,
KnowledgeBase knowledgeBase,
SemanticVerifier verifier,
MetadataProvider metadataProvider) {
this.llmClient = llmClient;
this.targetDialect = targetDialect;
this.knowledgeBase = knowledgeBase;
this.verifier = verifier;
this.metadataProvider = metadataProvider;
}

/**
【核心方法】使用 LLM 翻译 AST 规则引擎无法处理的 Phoenix 特性
- @param originalSql 原始 Phoenix SQL
  @param unresolvedFeatures AST 规则引擎未能解决的特征列表
  @param tableMetadata 相关表的元数据（列信息、PK、索引等）
  @return AI 翻译结果
- 处理流程（ReAct 模式）：
  构建 Prompt（包含上下文、知识库、约束）
  调用 LLM 生成目标方言 SQL
  验证生成的 SQL 语法正确性
  验证语义等价性（EXPLAIN 对比 + 数据集测试）
  如果验证失败，附加错误信息重新调用 LLM（最多 3 次）
  3 次都失败 → 标记为需要人工审核
  */
  public AiTranslationResult translate(String originalSql,
  List unresolvedFeatures,
  Map<String, TableMetadata> tableMetadata) {
  
  List attemptHistory = new ArrayList<>(); // 记录每次尝试的结果
  String lastError = null;
  
  for (int attempt = 1; attempt <= MAX_RETRIES; attempt++) {
  try {
  // ─── Step 1: 构建 Prompt ───
  String prompt = buildPrompt(originalSql, unresolvedFeatures,
  tableMetadata, attemptHistory, lastError);
```
     // ─── Step 2: 调用 LLM（带超时控制） ───
     CompletableFuture<String> future = CompletableFuture.supplyAsync(() ->
             llmClient.chat(prompt));

     String llmResponse;
     try {
         llmResponse = future.get(LLM_TIMEOUT_SECONDS, TimeUnit.SECONDS);
     } catch (TimeoutException e) {
         future.cancel(true);
         lastError = "LLM 调用超时（" + LLM_TIMEOUT_SECONDS + "s）";
         attemptHistory.add("Attempt " + attempt + ": " + lastError);
         continue;
     }

     // ─── Step 3: 解析 LLM 输出 ───
     // LLM 的输出格式约定：
     // sql
     // <翻译后的 SQL>
     //                 // <翻译说明>
     String translatedSql = extractSqlFromResponse(llmResponse);
     String explanation = extractExplanationFromResponse(llmResponse);

     if (translatedSql == null || translatedSql.isBlank()) {
         lastError = "LLM 输出中未找到有效的 SQL 代码块";
         attemptHistory.add("Attempt " + attempt + ": " + lastError);
         continue;
     }

     // ─── Step 4: 语法验证 ───
     // 使用目标方言的 Parser 验证生成的 SQL 是否语法正确
     SyntaxValidationResult syntaxResult = validateSyntax(translatedSql);
     if (!syntaxResult.isValid()) {
         lastError = "LLM 生成的 SQL 语法错误: " + syntaxResult.getErrorMessage();
         attemptHistory.add("Attempt " + attempt + ": " + lastError
                 "nGenerated SQL: " + translatedSql);
         continue;
     }

     // ─── Step 5: 语义等价验证 ───
     SemanticVerificationResult semanticResult = verifier.verify(
             originalSql, translatedSql, tableMetadata);

     if (semanticResult.isEquivalent()) {
         // ✅ 翻译成功！
         return AiTranslationResult.success(
                 originalSql, translatedSql, explanation,
                 semanticResult.getConfidence(), attempt,
                 semanticResult.getVerificationDetails()
         );
     } else {
         // ⚠️ 语义不等价，需要重新生成
         lastError = "语义等价验证失败: " + semanticResult.getDifferenceDescription();
         attemptHistory.add("Attempt " + attempt + ": " + lastError
                 "nGenerated SQL: " + translatedSql);
     }

 } catch (Exception e) {
     lastError = "异常: " + e.getMessage();
     attemptHistory.add("Attempt " + attempt + ": " + lastError);
 }
```
  }
  
  // ❌ 所有重试都失败，标记为需要人工审核
  return AiTranslationResult.manualReview(
  originalSql, unresolvedFeatures, attemptHistory, lastError);
  }
/**
【构建 Prompt】—— 这是整个 AI Agent 的灵魂
- Prompt 设计原则：
  角色设定：你是一个精通 Apache Phoenix 和 {目标方言} 的数据库迁移专家
  上下文注入：原始 SQL、Phoenix 特征标签、表元数据
  知识库增强：从 RAG 知识库中检索相似迁移案例
  约束声明：输出格式要求、禁止事项
  历史反馈：之前的尝试和错误信息（用于反思）
- @return 完整的 Prompt 文本
  */
  private String buildPrompt(String originalSql,
  List unresolvedFeatures,
  Map<String, TableMetadata> tableMetadata,
  List attemptHistory,
  String lastError) {
  StringBuilder prompt = new StringBuilder();
  
  // ─── System Prompt ───
  prompt.append(“”"
  你是一个专业的数据库 SQL 迁移引擎。你的任务是将 Apache Phoenix (HBase SQL 层) 的 SQL
  精确翻译为 %s 方言。
```
 【核心约束】
 语义必须 100%% 等价——不能改变查询的业务逻辑
 不能使用 %s 中不存在的函数或语法
 必须考虑性能——生成的 SQL 应该尽量利用索引
 如果某个 Phoenix 特性在 %s 中完全没有等价替换，请在注释中说明并提供最接近的替代方案
 保持原有的 Schema 前缀和表名不变
 
 【输出格式要求】
 你的回复必须严格遵循以下格式：
 
 sql
 <翻译后的完整 SQL>
             
 <翻译说明：逐条解释每个 Phoenix 特性是如何转换的>
 
 """.formatted(targetDialect, targetDialect, targetDialect));
```
  // ─── 知识库增强：检索相似迁移案例 ───
  List similarCases = knowledgeBase.search(
  originalSql, unresolvedFeatures, targetDialect, 3);
  
  if (!similarCases.isEmpty()) {
  prompt.append(“【参考案例】以下是类似的迁移案例，供你参考：nn”);
  for (int i = 0; i < similarCases.size(); i++) {
  KnowledgeEntry entry = similarCases.get(i);
  prompt.append(“案例 %d:n”.formatted(i + 1));
  prompt.append(" Phoenix SQL: “).append(entry.getPhoenixSql()).append(“n”);
  prompt.append(” %s SQL: “.formatted(targetDialect))
  .append(entry.getTargetSql()).append(“n”);
  prompt.append(” 说明: ").append(entry.getExplanation()).append(“nn”);
  }
  }
  
  // ─── 表元数据注入 ───
  prompt.append(“【表元数据】n”);
  for (Map.Entry<String, TableMetadata> entry : tableMetadata.entrySet()) {
  TableMetadata meta = entry.getValue();
  prompt.append("表名: “).append(entry.getKey()).append(“n”);
  prompt.append(” 列: “).append(meta.getColumnDescriptions()).append(“n”);
  prompt.append(” 主键: “).append(meta.getPrimaryKeyColumns()).append(“n”);
  prompt.append(” 索引: ").append(meta.getIndexDescriptions()).append(“nn”);
  }
  
  // ─── 原始 SQL 和待解决特征 ───
  prompt.append(“【待翻译的 Phoenix SQL】nsqln”)
  .append(originalSql).append(“nnn”);
  
  prompt.append(“【需要处理的 Phoenix 特有特性】n”);
  for (FeatureTag tag : unresolvedFeatures) {
  prompt.append("- “).append(tag.name())
  .append(”: ").append(tag.getDescription()).append(“n”);
  }
  
  // ─── 历史反馈（用于反思） ───
  if (!attemptHistory.isEmpty()) {
  prompt.append(“n【之前的尝试和失败原因】n”);
  prompt.append(“你已经尝试了 “).append(attemptHistory.size())
  .append(” 次，都失败了。请仔细分析失败原因，避免重复同样的错误。nn”);
  for (String history : attemptHistory) {
  prompt.append(history).append(“nn”);
  }
  if (lastError != null) {
  prompt.append("最后一次错误: ").append(lastError).append(“n”);
  }
  }
  
  return prompt.toString();
  }
/**
从 LLM 回复中提取 SQL 代码块
*/
private String extractSqlFromResponse(String response) {
// 匹配 sql … 格式
java.util.regex.Pattern pattern = java.util.regex.Pattern.compile(
“sql\n(.?)”,
java.util.regex.Pattern.DOTALL);
java.util.regex.Matcher matcher = pattern.matcher(response);
if (matcher.find()) {
return matcher.group(1).trim();
}
// 降级：匹配 … 格式
java.util.regex.Pattern fallback = java.util.regex.Pattern.compile(
“\n(.?)”,
java.util.regex.Pattern.DOTALL);
matcher = fallback.matcher(response);
if (matcher.find()) {
return matcher.group(1).trim();
}
return null;
}

/**
从 LLM 回复中提取翻译说明
*/
private String extractExplanationFromResponse(String response) {
// 提取之后的所有文本作为说明
int lastCodeBlock = response.lastIndexOf(“”);
if (lastCodeBlock >= 0 && lastCodeBlock + 3 < response.length()) {
return response.substring(lastCodeBlock + 3).trim();
}
return “”;
}

/**
语法验证
*/
private SyntaxValidationResult validateSyntax(String translatedSql) {
try {
// 使用目标方言的 Parser 尝试解析
// 如果解析成功，说明语法正确
switch (targetDialect) {
case DM:
// 使用 JSqlParser 解析（达梦兼容 Oracle 语法）
net.sf.jsqlparser.parser.CCJSqlParserUtil.parse(translatedSql);
break;
case KINGBASE:
// 金仓兼容 PostgreSQL 语法
net.sf.jsqlparser.parser.CCJSqlParserUtil.parse(translatedSql);
break;
case OCEANBASE_MYSQL:
case TIDB:
// MySQL 兼容语法
net.sf.jsqlparser.parser.CCJSqlParserUtil.parse(translatedSql);
break;
}
return SyntaxValidationResult.valid();
} catch (Exception e) {
return SyntaxValidationResult.invalid(e.getMessage());
}
}
}

/**
AI 翻译结果
*/
public class AiTranslationResult {

public enum Status {
    SUCCESS,        // 翻译成功且通过验证
    MANUAL_REVIEW   // 需要人工审核
}

private final Status status;
private final String originalSql;
private final String translatedSql;       // SUCCESS 时有值
private final String explanation;         // 翻译说明
private final double confidence;          // 置信度 (0.0 - 1.0)
private final int attempts;               // 尝试次数
private final List<FeatureTag> unresolvedFeatures; // MANUAL_REVIEW 时有值
private final List<String> attemptHistory;
private final String lastError;
private final String verificationDetails;

public static AiTranslationResult success(String original, String translated,
                                            String explanation, double confidence,
                                            int attempts, String verificationDetails) {
    return new AiTranslationResult(Status.SUCCESS, original, translated,
            explanation, confidence, attempts, null, null, null, verificationDetails);
}

public static AiTranslationResult manualReview(String original,
                                                 List<FeatureTag> unresolved,
                                                 List<String> history,
                                                 String lastError) {
    return new AiTranslationResult(Status.MANUAL_REVIEW, original, null,
            null, 0.0, history.size(), unresolved, history, lastError, null);
}

// 私有构造器 + Getter 省略
private AiTranslationResult(Status status, String originalSql, String translatedSql,
                              String explanation, double confidence, int attempts,
                              List<FeatureTag> unresolvedFeatures,
                              List<String> attemptHistory, String lastError,
                              String verificationDetails) {
    this.status = status;
    this.originalSql = originalSql;
    this.translatedSql = translatedSql;
    this.explanation = explanation;
    this.confidence = confidence;
    this.attempts = attempts;
    this.unresolvedFeatures = unresolvedFeatures;
    this.attemptHistory = attemptHistory;
    this.lastError = lastError;
    this.verificationDetails = verificationDetails;
}

public Status getStatus() { return status; }
public String getOriginalSql() { return originalSql; }
public String getTranslatedSql() { return translatedSql; }
public String getExplanation() { return explanation; }
public double getConfidence() { return confidence; }
public int getAttempts() { return attempts; }
public List<FeatureTag> getUnresolvedFeatures() { return unresolvedFeatures; }
public List<String> getAttemptHistory() { return attemptHistory; }
public String getLastError() { return lastError; }

}

5.2 知识库 RAG 系统（迁移经验沉淀）

package com.dragonmigrate.phoenix.ai;

import com.dragonmigrate.core.TargetDialect;
import com.dragonmigrate.phoenix.analysis.PhoenixFeatureExtractor.FeatureTag;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;

/**

🟣 Phoenix 迁移知识库 (Migration Knowledge Base with RAG)

【设计思想】
LLM 的"幻觉"问题可以通过 RAG (Retrieval-Augmented Generation) 大幅缓解。
知识库包含两类知识：
专家规则（Expert Rules）：
人工编写的 Phoenix→目标方言的映射规则，100% 正确
例如：“Phoenix SALT_BUCKETS=N → 达梦 HASH 分区表（N个分区）”
1. 历史案例（Historical Cases）：
  之前成功迁移的 SQL 对（原始SQL + 翻译后SQL + 人工确认）
  这些案例会被向量化存储，供相似性检索
【向量化检索】

SQL 文本 → Embedding 向量（使用 text-embedding-ada-002 或本地模型）
特征标签 → 稀疏向量（47维，one-hot编码）
检索时：混合相似度 = 0.6 * text_similarity + 0.4 * feature_similarity

【知识积累】
每次人工确认的翻译结果，会自动加入知识库
→ 系统越用越聪明（飞轮效应）
*/
public class KnowledgeBase {

/** 专家规则库 */
private final Map<String, List> expertRules;

/** 历史案例库（向量索引） */
private final VectorIndex caseIndex;

/** Embedding 模型客户端 */
private final EmbeddingClient embeddingClient;

public KnowledgeBase(EmbeddingClient embeddingClient) {
this.embeddingClient = embeddingClient;
this.expertRules = new ConcurrentHashMap<>();
this.caseIndex = new InMemoryVectorIndex();
initializeExpertRules();
}

/**
【初始化专家规则】
预置的 Phoenix → 国产库映射规则
*/
private void initializeExpertRules() {
// ─── SALT_BUCKETS 迁移规则 ───
addExpertRule(“DM”, new KnowledgeEntry(
“CREATE TABLE t (…) SALT_BUCKETS=16”,
“”"
CREATE TABLE t (…)
PARTITION BY HASH(primary_key_column)
PARTITIONS 16
“”“,
“””
Phoenix SALT_BUCKETS 的作用是在 HBase 中对 RowKey 做 Hash 分散，
避免热点 Region。在达梦中，最接近的等价方案是 HASH 分区表。
分区数设为与 SALT_BUCKETS 相同的值。
注意：需要确认主键列名，替换 primary_key_column。
“”",
List.of(FeatureTag.SALT_BUCKETS),
TargetDialect.DM,
1.0 // 置信度
));

 addExpertRule("KINGBASE", new KnowledgeEntry(
         "CREATE TABLE t (...) SALT_BUCKETS=16",
         """
         CREATE TABLE t (...)
         PARTITION BY HASH(primary_key_column)
         PARTITIONS 16
         """,
         "金仓支持 HASH 分区，语义与达梦相同。",
         List.of(FeatureTag.SALT_BUCKETS),
         TargetDialect.KINGBASE,
         1.0
 ));

 // ─── IMMUTABLE_ROWS 迁移规则 ───
 addExpertRule("DM", new KnowledgeEntry(
         "CREATE TABLE t (...) IMMUTABLE_ROWS=true",
         """
         CREATE TABLE t (...)
         /* 注意：原表标记为 IMMUTABLE_ROWS，表示数据写入后不会被 UPDATE。
            建议在应用层保证此约束。如果需要数据库层面的保护，
            可以添加以下触发器： */;
         
         CREATE OR REPLACE TRIGGER trg_immutable_guard
         BEFORE UPDATE ON t
         FOR EACH ROW
         BEGIN
             RAISE_APPLICATION_ERROR(-20001, 
                 '此表为 IMMUTABLE 表，禁止 UPDATE 操作');
         END;
         """,
         """
         Phoenix IMMUTABLE_ROWS 告诉 HBase 该表的数据写入后不会修改，
         HBase 据此做存储优化（不维护多版本）。
         在关系库中没有直接等价特性。
         最佳实践：
         在文档中标注此表为"只追加"表
         可选：添加触发器阻止 UPDATE/DELETE
         """,
         List.of(FeatureTag.IMMUTABLE_ROWS),
         TargetDialect.DM,
         0.85
 ));

 // ─── 列族展平规则 ───
 addExpertRule("ALL", new KnowledgeEntry(
         "CREATE TABLE t (cf1.col1 VARCHAR, cf2.col2 INTEGER)",
         "CREATE TABLE t (col1 VARCHAR, col2 INTEGER)",
         """
         Phoenix 列族是 HBase 的物理存储概念（同一列族的数据存储在同一 StoreFile 中）。
         在关系库中，所有列存储在同一行中，列族概念无意义。
         迁移方案：移除列族前缀，将 cf.column 展平为 column。
         ⚠️ 注意：如果列名在不同列族下有重名（如 cf1.name 和 cf2.name），
         需要重命名以避免冲突（如 name_1, name_2）。
         """,
         List.of(FeatureTag.COLUMN_FAMILY),
         null, // 适用于所有方言
         0.95
 ));

 // ─── GLOBAL INDEX 迁移规则 ───
 addExpertRule("DM", new KnowledgeEntry(
         "CREATE INDEX idx ON t (col1, col2 DESC) INCLUDE (col3, col4)",
         """
         -- Phoenix Global Index 在 HBase 中是一张独立的索引表
         -- 在达梦中等价于普通 B-Tree 索引 + 附加列
         CREATE INDEX idx ON t (col1, col2 DESC);
         -- 注意：达梦不支持 INCLUDE 语法，但覆盖索引的效果可以通过
         -- 将所有需要的列放入索引来实现：
         -- CREATE INDEX idx ON t (col1, col2 DESC, col3, col4);
         -- 需要根据实际查询模式决定是否扩展索引列
         """,
         """
         Phoenix Global Index 底层是独立的 HBase 表，支持跨 Region 的索引查找。
         在达梦中，普通索引已经覆盖了此功能。
         关键差异：Phoenix 的 INCLUDE 子句（覆盖索引附加列）在达梦中不直接支持。
         替代方案：
         将 INCLUDE 列也加入索引定义（变成联合索引）
         或依赖达梦的"索引快速全扫描"特性
         选择取决于查询频率和索引大小。
         """,
         List.of(FeatureTag.GLOBAL_INDEX, FeatureTag.INDEX_INCLUDE),
         TargetDialect.DM,
         0.9
 ));

 // ─── LOCAL INDEX 迁移规则 ───
 addExpertRule("DM", new KnowledgeEntry(
         "CREATE LOCAL INDEX idx ON t (col1)",
         "CREATE INDEX idx ON t (col1)",
         """
         Phoenix Local Index 与数据共处一个 Region，适合范围扫描。
         在达梦中，如果表是分区表，可以创建 LOCAL 索引：
         CREATE INDEX idx ON t (col1) LOCAL;
         如果表不是分区表，普通索引即可。
         """,
         List.of(FeatureTag.LOCAL_INDEX),
         TargetDialect.DM,
         0.85
 ));

 // ─── DELETE...LIMIT 迁移规则 ───
 addExpertRule("DM", new KnowledgeEntry(
         "DELETE FROM t WHERE condition LIMIT 1000",
         """
         DELETE FROM t 
         WHERE ROWID IN (
             SELECT ROWID FROM t 
             WHERE condition 
             AND ROWNUM <= 1000
         )
         """,
         """
         Phoenix DELETE 支持 LIMIT 子句（限制删除行数），标准 SQL 不支持。
         达梦替代方案：通过 ROWNUM 子查询限制行数。
         金仓替代方案：使用 ctid（类似 ROWID）+ LIMIT。
         """,
         List.of(FeatureTag.DELETE_LIMIT),
         TargetDialect.DM,
         0.95
 ));

 addExpertRule("KINGBASE", new KnowledgeEntry(
         "DELETE FROM t WHERE condition LIMIT 1000",
         """
         DELETE FROM t 
         WHERE ctid IN (
             SELECT ctid FROM t 
             WHERE condition 
             LIMIT 1000
         )
         """,
         "金仓使用 ctid（PostgreSQL 系统列）替代 ROWID。",
         List.of(FeatureTag.DELETE_LIMIT),
         TargetDialect.KINGBASE,
         0.95
 ));

}

/**
【检索相似案例】混合相似度检索
*/
public List search(String querySql,
List queryFeatures,
TargetDialect targetDialect,
int topK) {
// 1. 文本向量相似度检索
float[] queryEmbedding = embeddingClient.embed(querySql);
List textResults = caseIndex.search(queryEmbedding, topK * 2);

 // 2. 特征标签匹配度计算
 Set<FeatureTag> queryTagSet = EnumSet.copyOf(queryFeatures);
 
 // 3. 混合排序
 List<KnowledgeEntry> candidates = new ArrayList<>();
 for (VectorSearchResult result : textResults) {
     KnowledgeEntry entry = (KnowledgeEntry) result.getPayload();
     
     // 方言过滤
     if (entry.getTargetDialect() != null && 
         entry.getTargetDialect() != targetDialect) {
         continue;
     }

     // 混合相似度
     double textSim = result.getScore();
     double featureSim = computeFeatureSimilarity(queryTagSet, entry.getFeatures());
     double mixedScore = 0.6 * textSim + 0.4 * featureSim;

     entry.setSearchScore(mixedScore);
     candidates.add(entry);
 }

 // 4. 补充专家规则（精确匹配特征标签）
 for (FeatureTag tag : queryFeatures) {
     String ruleKey = targetDialect.name();
     List<KnowledgeEntry> rules = expertRules.get(ruleKey);
     if (rules != null) {
         for (KnowledgeEntry rule : rules) {
             if (rule.getFeatures().contains(tag)) {
                 rule.setSearchScore(1.0); // 专家规则最高优先级
                 candidates.add(rule);
             }
         }
     }
     // 通用规则
     List<KnowledgeEntry> genericRules = expertRules.get("ALL");
     if (genericRules != null) {
         for (KnowledgeEntry rule : genericRules) {
             if (rule.getFeatures().contains(tag)) {
                 rule.setSearchScore(0.95);
                 candidates.add(rule);
             }
         }
     }
 }

 // 去重 + 排序 + 取 TopK
 return candidates.stream()
         .collect(java.util.stream.Collectors.toMap(
                 KnowledgeEntry::getPhoenixSql,
                 e -> e,
                 (a, b) -> a.getSearchScore() >= b.getSearchScore() ? a : b))
         .values().stream()
         .sorted(Comparator.comparingDouble(KnowledgeEntry::getSearchScore).reversed())
         .limit(topK)
         .collect(java.util.stream.Collectors.toList());

}

private double computeFeatureSimilarity(Set query,
Set candidate) {
if (query.isEmpty() && candidate.isEmpty()) return 1.0;
if (query.isEmpty() || candidate.isEmpty()) return 0.0;

 Set<FeatureTag> intersection = EnumSet.copyOf(query);
 intersection.retainAll(candidate);

 Set<FeatureTag> union = EnumSet.copyOf(query);
 union.addAll(candidate);

 return (double) intersection.size() / union.size(); // Jaccard 相似度

}

private void addExpertRule(String dialectKey, KnowledgeEntry entry) {
expertRules.computeIfAbsent(dialectKey, k -> new ArrayList<>()).add(entry);
}
}

六、核心模块四：语义等价验证框架（质量守门员 ⭐⭐⭐⭐⭐）

这是整个引擎的质量底线。无论 AI 翻译得多"像"，都必须通过验证才能上线。

package com.dragonmigrate.phoenix.verification;

import com.dragonmigrate.core.TargetDialect;
import java.util.*;
import java.math.BigDecimal;

/**

🔴 语义等价验证框架 (Semantic Equivalence Verification Framework)

【设计思想】
验证"两条 SQL 语义是否等价"本质上是一个不可判定问题（等价于程序等价性验证）。
但在工程实践中，我们可以通过以下"组合验证"策略达到 99.9% 的置信度：
验证层级（由浅入深）：
Level 1: 结构验证 (Structural Verification)

检查 SELECT 的列名/列数是否一致
检查 WHERE 条件的过滤逻辑是否等价
检查 JOIN 关系是否保留
→ 成本：极低（纯 AST 比对），可以发现 50% 的错误

Level 2: 执行计划验证 (EXPLAIN Plan Verification)

对比两条 SQL 的 EXPLAIN 输出
检查扫描方式（全表扫描 vs 索引扫描）是否一致
检查 JOIN 顺序是否合理
→ 成本：中等（需要连接数据库），可以发现性能退化

Level 3: 数据集验证 (Dataset Verification)

使用标准测试数据集（包含各种边界条件）
分别在源库和目标库上执行两条 SQL
对比结果集是否完全一致（行数、列值、顺序）
→ 成本：较高（需要实际执行），可以发现 99% 的语义错误

Level 4: 模糊验证 (Fuzzy Verification)

对于无法精确对比的情况（如浮点数精度差异）
使用"容差比较"（差值 < epsilon）
检查统计特征（SUM/AVG/COUNT/MIN/MAX）是否一致
→ 成本：中等，补充 Level 3 的精度差异

⚠️ 重要：
任何单一验证层级都不能保证 100% 正确。
必须四层全部通过，才能标记为 EQUIVALENT。
@author 墨夶
@since 2026-07-01
*/
public class SemanticVerifier {

private final StructuralVerifier structuralVerifier;
private final ExplainPlanVerifier explainVerifier;
private final DatasetVerifier datasetVerifier;
private final FuzzyVerifier fuzzyVerifier;

public SemanticVerifier(StructuralVerifier sv, ExplainPlanVerifier ev,
DatasetVerifier dv, FuzzyVerifier fv) {
this.structuralVerifier = sv;
this.explainVerifier = ev;
this.datasetVerifier = dv;
this.fuzzyVerifier = fv;
}

/**
【核心验证方法】四层组合验证
- @param phoenixSql 原始 Phoenix SQL
  @param translatedSql 翻译后的目标方言 SQL
  @param tableMetadata 相关表的元数据
  @return 语义等价验证结果
  */
  public SemanticVerificationResult verify(String phoenixSql,
  String translatedSql,
  Map<String, TableMetadata> tableMetadata) {
  List details = new ArrayList<>();
  double totalConfidence = 1.0;
  
  // ─── Level 1: 结构验证 ───
  StructuralResult l1 = structuralVerifier.verify(phoenixSql, translatedSql);
  details.add(new VerificationDetail(1, “结构验证”,
  l1.isPassed(), l1.getDescription(), l1.getConfidence()));
  
  if (!l1.isPassed()) {
  // Level 1 失败 → 直接判定不等价（无需继续验证）
  return SemanticVerificationResult.notEquivalent(
  details, l1.getDescription(), computeConfidence(details));
  }
  totalConfidence *= l1.getConfidence();
  
  // ─── Level 2: 执行计划验证（可选，需要数据库连接） ───
  try {
  ExplainResult l2 = explainVerifier.verify(phoenixSql, translatedSql, tableMetadata);
  details.add(new VerificationDetail(2, “执行计划验证”,
  l2.isPassed(), l2.getDescription(), l2.getConfidence()));
  totalConfidence *= l2.getConfidence();
  } catch (Exception e) {
  // EXPLAIN 验证失败（可能没有数据库连接），跳过但不影响整体判断
  details.add(new VerificationDetail(2, “执行计划验证”,
  true, "跳过（无数据库连接）: " + e.getMessage(), 0.9));
  totalConfidence *= 0.9;
  }
  
  // ─── Level 3: 数据集验证 ───
  try {
  DatasetResult l3 = datasetVerifier.verify(phoenixSql, translatedSql, tableMetadata);
  details.add(new VerificationDetail(3, “数据集验证”,
  l3.isPassed(), l3.getDescription(), l3.getConfidence()));
```
 if (!l3.isPassed()) {
     return SemanticVerificationResult.notEquivalent(
             details, "数据集验证失败: " + l3.getDifferenceDescription(),
             computeConfidence(details));
 }
 totalConfidence *= l3.getConfidence();
```
  } catch (Exception e) {
  details.add(new VerificationDetail(3, “数据集验证”,
  true, "跳过: " + e.getMessage(), 0.85));
  totalConfidence *= 0.85;
  }
  
  // ─── Level 4: 模糊验证 ───
  FuzzyResult l4 = fuzzyVerifier.verify(phoenixSql, translatedSql, tableMetadata);
  details.add(new VerificationDetail(4, “模糊验证”,
  l4.isPassed(), l4.getDescription(), l4.getConfidence()));
  totalConfidence *= l4.getConfidence();
  
  // 综合判定
  double finalConfidence = computeConfidence(details);
  if (finalConfidence >= 0.95) {
  return SemanticVerificationResult.equivalent(details, finalConfidence);
  } else if (finalConfidence >= 0.80) {
  return SemanticVerificationResult.probablyEquivalent(details, finalConfidence);
  } else {
  return SemanticVerificationResult.notEquivalent(
  details, "综合置信度过低: " + finalConfidence, finalConfidence);
  }
  }
private double computeConfidence(List details) {
return details.stream()
.mapToDouble(VerificationDetail::getConfidence)
.reduce(1.0, (a, b) -> a * b);
}
}

/**
Level 1: 结构验证器
*/
public class StructuralVerifier {

/**
 【结构验证】比对两条 SQL 的"骨架"是否一致
 * 验证项目：
 SELECT 输出列数量和别名是否一致
 FROM 中引用的表集合是否一致（忽略顺序）
 JOIN 类型和 JOIN 条件是否等价
 WHERE 条件中的过滤列是否一致
 GROUP BY 列是否一致
 ORDER BY 列和排序方向是否一致
 HAVING 条件是否一致
 聚合函数是否一致（SUM/COUNT/AVG/MIN/MAX）
 * ⚠️ 注意：
 - Phoenix 的 ARRAY_ELEM(arr, 1) 和金仓的 arr[1] 在结构上"不同"
 但语义相同。结构验证器应该忽略这种"语法差异但语义等价"的情况。
 - 解决方案：在结构比对之前，先做一层"函数归一化"
 将所有方言的等价函数映射到统一的"规范函数名"
 */
public StructuralResult verify(String phoenixSql, String translatedSql) {
    // 1. 解析两条 SQL 的 AST
    SqlStructure phoenixStructure = extractStructure(phoenixSql, "PHOENIX");
    SqlStructure translatedStructure = extractStructure(translatedSql, "TARGET");

    List<String> differences = new ArrayList<>();

    // 2. 比对 SELECT 列
    if (phoenixStructure.getSelectColumnCount() != translatedStructure.getSelectColumnCount()) {
        differences.add("SELECT 列数不同: Phoenix=" + 
                phoenixStructure.getSelectColumnCount() + 
                ", Target=" + translatedStructure.getSelectColumnCount());
    }

    // 比对列别名
    List<String> phoenixAliases = phoenixStructure.getSelectAliases();
    List<String> translatedAliases = translatedStructure.getSelectAliases();
    if (!phoenixAliases.equals(translatedAliases)) {
        differences.add("SELECT 列别名不同: " + phoenixAliases + " vs " + translatedAliases);
    }

    // 3. 比对 FROM 表集合
    Set<String> phoenixTables = new TreeSet<>(phoenixStructure.getFromTables());
    Set<String> translatedTables = new TreeSet<>(translatedStructure.getFromTables());
    if (!phoenixTables.equals(translatedTables)) {
        differences.add("FROM 表集合不同: " + phoenixTables + " vs " + translatedTables);
    }

    // 4. 比对 JOIN 类型
    if (!phoenixStructure.getJoinTypes().equals(translatedStructure.getJoinTypes())) {
        differences.add("JOIN 类型不同: " + 
                phoenixStructure.getJoinTypes() + " vs " + translatedStructure.getJoinTypes());
    }

    // 5. 比对 GROUP BY
    if (!phoenixStructure.getGroupByColumns().equals(translatedStructure.getGroupByColumns())) {
        differences.add("GROUP BY 列不同");
    }

    // 6. 比对 ORDER BY
    if (!phoenixStructure.getOrderByColumns().equals(translatedStructure.getOrderByColumns())) {
        differences.add("ORDER BY 列或方向不同");
    }

    // 7. 比对聚合函数
    Set<String> phoenixAggs = normalizeAggregates(phoenixStructure.getAggregateFunctions());
    Set<String> translatedAggs = normalizeAggregates(translatedStructure.getAggregateFunctions());
    if (!phoenixAggs.equals(translatedAggs)) {
        differences.add("聚合函数不同: " + phoenixAggs + " vs " + translatedAggs);
    }

    if (differences.isEmpty()) {
        return StructuralResult.passed("结构验证通过", 0.7);
    } else {
        return StructuralResult.failed(String.join("; ", differences));
    }
}

/**
 聚合函数归一化
 将不同方言的等价聚合函数映射到统一名称
 */
private Set<String> normalizeAggregates(Set<String> aggregates) {
    Set<String> normalized = new TreeSet<>();
    for (String agg : aggregates) {
        String upper = agg.toUpperCase();
        // 归一化映射
        if (upper.contains("COUNT")) normalized.add("COUNT");
        else if (upper.contains("SUM")) normalized.add("SUM");
        else if (upper.contains("AVG")) normalized.add("AVG");
        else if (upper.contains("MIN")) normalized.add("MIN");
        else if (upper.contains("MAX")) normalized.add("MAX");
        else if (upper.contains("LISTAGG") || upper.contains("GROUP_CONCAT") 
                || upper.contains("STRING_AGG")) normalized.add("STRING_AGG");
        else normalized.add(upper);
    }
    return normalized;
}

private SqlStructure extractStructure(String sql, String dialect) {
    // 使用 JSqlParser 提取 SQL 结构信息
    // 省略具体实现...
    return new SqlStructure();
}

}

/**
Level 3: 数据集验证器
*/
public class DatasetVerifier {

/**
 【数据集验证】使用标准测试数据集执行两条 SQL 并对比结果
 * 测试数据集设计原则：
 覆盖正常数据（典型业务场景）
 覆盖边界数据（NULL值、空字符串、极值、重复值）
 覆盖特殊字符（Unicode、Emoji、SQL 注入字符）
 数据量适中（1000-10000行，保证验证速度）
 * 比对策略：
 行数比对（COUNT）
 列值逐行比对（考虑排序）
 统计特征比对（SUM/AVG/MIN/MAX）
 NULL 值分布比对
 */
public DatasetResult verify(String phoenixSql, String translatedSql,
                              Map<String, TableMetadata> tableMetadata) {
    // 1. 在测试数据集上执行 Phoenix SQL（通过 Phoenix 测试连接）
    // 2. 在测试数据集上执行翻译后 SQL（通过目标库测试连接）
    // 3. 对比结果集

    // 简化示例：
    // ResultSet phoenixResult = phoenixConnection.executeQuery(phoenixSql);
    // ResultSet targetResult = targetConnection.executeQuery(translatedSql);
    // return compareResultSets(phoenixResult, targetResult);

    return DatasetResult.passed("数据集验证通过（标准测试集 1000 行，结果一致）", 0.95);
}

}

七、迁移编排引擎：3200 条 SQL 的流水线调度

package com.dragonmigrate.phoenix.orchestration;

import com.dragonmigrate.core.TargetDialect;
import com.dragonmigrate.phoenix.analysis.;
import com.dragonmigrate.phoenix.mutation.;
import com.dragonmigrate.phoenix.ai.;
import com.dragonmigrate.phoenix.verification.;
import java.util.;
import java.util.concurrent.;
import java.util.concurrent.atomic.;
import java.nio.file.;
import java.io.*;

/**

🔴 批量迁移编排引擎 (Batch Migration Orchestrator)

将 3200 条 Phoenix SQL 的迁移任务，编排为一条完整的流水线：
┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐
│ SQL 采集 │───▶│ 特征分析 │───▶│ 自动迁移 │───▶│ AI 迁移 │───▶│ 质量验证 │
│ (Collect) │ │ (Analyze) │ │ (AST) │ │ (LLM) │ │ (Verify) │
└──────────┘ └──────────┘ └──────────┘ └──────────┘ └──────────┘
│ │ │ │ │
▼ ▼ ▼ ▼ ▼
┌─────────────────────────────────────────────────────────────────────────┐
│ 迁移报告生成 (Report Generator) │
└─────────────────────────────────────────────────────────────────────────┘
*/
public class MigrationOrchestrator {

private final PhoenixFeatureExtractor featureExtractor;
private final AstMutationEngine mutationEngine;
private final LlmdialectTranslationAgent aiAgent;
private final SemanticVerifier verifier;
private final TargetDialect targetDialect;

/** 并发度（CPU 密集型任务，设为 CPU 核数） */
private final int parallelism;

/** 全局统计 */
private final AtomicLong totalSqlCount = new AtomicLong(0);
private final AtomicLong passthroughCount = new AtomicLong(0);
private final AtomicLong autoMigratedCount = new AtomicLong(0);
private final AtomicLong aiMigratedCount = new AtomicLong(0);
private final AtomicLong manualReviewCount = new AtomicLong(0);
private final AtomicLong failedCount = new AtomicLong(0);

public MigrationOrchestrator(TargetDialect targetDialect, int parallelism) {
this.targetDialect = targetDialect;
this.parallelism = parallelism;
```
 // 初始化各模块
 this.featureExtractor = new PhoenixFeatureExtractor();
 this.mutationEngine = new AstMutationEngine(targetDialect);
 this.aiAgent = new LlmdialectTranslationAgent(
         new OpenAiClient(System.getenv("OPENAI_API_KEY")),
         targetDialect,
         new KnowledgeBase(new EmbeddingClient()),
         new SemanticVerifier(/* ... */),
         new JdbcMetadataProvider(/* ... */)
 );
 this.verifier = new SemanticVerifier(/* ... */);
```
}

/**
【核心流水线】批量迁移
- @param sqlSources SQL 来源列表（文件路径、代码片段、MyBatis XML 等）
  @return 完整迁移报告
- 🚀 性能目标：
- 3200 条 SQL 全量迁移 < 2 小时
- 其中 PASSTHROUGH（直接通过）约 60% → 秒级
- AUTO_AST（自动变换）约 25% → 分钟级
- AI_ASSISTED（AI辅助）约 10% → 小时级（受 LLM API 限制）
- MANUAL（人工）约 5% → 提交人工审核队列
  */
  public MigrationReport migrate(List sqlSources) {
  long startTime = System.currentTimeMillis();
  
  // ─── Phase 1: SQL 采集与解析 ───
  System.out.println(“📥 Phase 1: SQL 采集…”);
  List rawSqlList = collectAllSql(sqlSources);
  totalSqlCount.set(rawSqlList.size());
  System.out.println(" 采集到 " + rawSqlList.size() + " 条 SQL");
  
  // ─── Phase 2: 特征分析（并行） ───
  System.out.println(“🔍 Phase 2: 特征分析…”);
  ExecutorService analysisPool = Executors.newFixedThreadPool(parallelism);
  List<Future> analysisResults = new ArrayList<>();
  
  for (RawSql raw : rawSqlList) {
  analysisResults.add(analysisPool.submit(() -> {
  SqlFingerprint fingerprint = featureExtractor.extract(raw.getSql());
  return new AnalyzedSql(raw, fingerprint);
  }));
  }
  
  List analyzedList = new ArrayList<>();
  for (Future future : analysisResults) {
  try {
  analyzedList.add(future.get());
  } catch (Exception e) {
  System.err.println(" ⚠️ 特征分析失败: " + e.getMessage());
  failedCount.incrementAndGet();
  }
  }
  analysisPool.shutdown();
  
  // 统计各迁移路径的分布
  Map<PhoenixFeatureExtractor.MigrationPath, Long> pathDistribution = analyzedList.stream()
  .collect(java.util.stream.Collectors.groupingBy(
  a -> a.getFingerprint().getRecommendedPath(),
  java.util.stream.Collectors.counting()));
  
  System.out.println(" 迁移路径分布:“);
  pathDistribution.forEach((path, count) ->
  System.out.printf(” %s: %d 条 (%.1f%%)%n",
  path.getLabel(), count, count * 100.0 / totalSqlCount.get()));
  
  // ─── Phase 3: 自动迁移（AST 规则引擎） ───
  System.out.println(“⚙️ Phase 3: AST 自动迁移…”);
  List results = Collections.synchronizedList(new ArrayList<>());
  
  ExecutorService mutationPool = Executors.newFixedThreadPool(parallelism);
  List needsAi = new ArrayList<>(); // 需要 AI 处理的 SQL
  List needsManual = new ArrayList<>(); // 需要人工的 SQL
  
  for (AnalyzedSql analyzed : analyzedList) {
  PhoenixFeatureExtractor.MigrationPath path = analyzed.getFingerprint().getRecommendedPath();
```
 switch (path) {
     case PASSTHROUGH:
         // 无需迁移
         results.add(MigrationResult.passthrough(analyzed));
         passthroughCount.incrementAndGet();
         break;

     case AUTO_AST:
     case SEMI_AUTO:
         // AST 规则引擎处理
         mutationPool.submit(() -> {
             try {
                 MutationResult mutation = mutationEngine.mutate(
                         analyzed.getAst(), analyzed.getFingerprint());
                 
                 if (mutation.getUnresolvedTags().isEmpty()) {
                     // AST 规则引擎解决了所有问题
                     results.add(MigrationResult.autoSuccess(analyzed, mutation));
                     autoMigratedCount.incrementAndGet();
                 } else {
                     // 还有未解决的特征，需要 AI 补充处理
                     synchronized (needsAi) {
                         needsAi.add(analyzed.withPartialMutation(mutation));
                     }
                 }
             } catch (Exception e) {
                 // AST 规则失败，降级到 AI
                 synchronized (needsAi) {
                     needsAi.add(analyzed);
                 }
             }
         });
         break;

     case AI_ASSISTED:
         needsAi.add(analyzed);
         break;

     case MANUAL:
         needsManual.add(analyzed);
         manualReviewCount.incrementAndGet();
         results.add(MigrationResult.manualRequired(analyzed));
         break;
 }
```
  }
  mutationPool.shutdown();
  try { mutationPool.awaitTermination(30, TimeUnit.MINUTES); } catch (InterruptedException e) { /* … */ }
  
  // ─── Phase 4: AI 迁移（串行或低并发，受 LLM API 限制） ───
  System.out.println(“🤖 Phase 4: AI 辅助迁移 (” + needsAi.size() + " 条)…");
  ExecutorService aiPool = Executors.newFixedThreadPool(
  Math.min(4, parallelism)); // LLM API 并发限制
  
  for (AnalyzedSql sql : needsAi) {
  aiPool.submit(() -> {
  try {
  AiTranslationResult aiResult = aiAgent.translate(
  sql.getSql(),
  new ArrayList<>(sql.getFingerprint().getDetectedTags()),
  sql.getTableMetadata()
  );
```
         if (aiResult.getStatus() == AiTranslationResult.Status.SUCCESS) {
             results.add(MigrationResult.aiSuccess(sql, aiResult));
             aiMigratedCount.incrementAndGet();
         } else {
             results.add(MigrationResult.manualRequired(sql));
             manualReviewCount.incrementAndGet();
         }
     } catch (Exception e) {
         results.add(MigrationResult.failed(sql, e.getMessage()));
         failedCount.incrementAndGet();
     }
 });
```
  }
  aiPool.shutdown();
  try { aiPool.awaitTermination(2, TimeUnit.HOURS); } catch (InterruptedException e) { /* … */ }
  
  // ─── Phase 5: 生成迁移报告 ───
  long elapsed = System.currentTimeMillis() - startTime;
  MigrationReport report = new MigrationReport(
  targetDialect,
  totalSqlCount.get(),
  passthroughCount.get(),
  autoMigratedCount.get(),
  aiMigratedCount.get(),
  manualReviewCount.get(),
  failedCount.get(),
  elapsed,
  results,
  pathDistribution
  );
  
  // 输出报告到文件
  writeReport(report, Path.of(“migration_report_” + targetDialect.name() + “.json”));
  writeSqlMapping(results, Path.of(“sql_mapping_” + targetDialect.name() + “.csv”));
  
  return report;
  }
/**
SQL 采集：从多种来源提取 SQL
*/
private List collectAllSql(List sources) {
List allSql = new ArrayList<>();
for (SqlSource source : sources) {
allSql.addAll(source.extract());
}
return allSql;
}

private void writeReport(MigrationReport report, Path path) {
// JSON 序列化输出
// 省略…
}

private void writeSqlMapping(List results, Path path) {
// CSV 格式输出映射关系：原始SQL, 翻译后SQL, 迁移路径, 置信度
// 省略…
}
}

八、生产实战：72 小时迁移实录与 7 条血泪铁律

8.1 真实迁移时间线
时间阶段成果
Day 1 AM SQL 采集从 Java 代码、MyBatis XML、HBase Shell 脚本中采集到 3247 条 SQL

Day 1 PM 特征分析 62% 标准SQL（PASSTHROUGH），23% AUTO_AST，11% AI_ASSISTED，4% MANUAL

Day 2 AM AST 自动迁移 746 条 SQL 全部通过确定性变换，零错误

Day 2 PM AI 辅助迁移 357 条 SQL，AI 成功 312 条（87%），45 条降级为人工

Day 3 AM 人工审核 45 条 AI 失败 + 130 条 MANUAL = 175 条，3 名 DBA 并行处理

Day 3 PM 全量验证四层验证通过，生成最终迁移报告

最终数据：
✅ 自动迁移率：94.6%（3072/3247）
⚠️ 人工介入率：5.4%（175/3247）
🎯 语义等价验证通过率：99.7%
⏱️ 总耗时：72 小时（传统人工估计 1280 小时）

8.2 七条血泪铁律

铁律 1：永远不要让 LLM 直接翻译整条 SQL

翻车案例：让 LLM 翻译一条 200 行的嵌套 CTE 查询，LLM 把中间一个子查询的 LEFT JOIN 改成了 INNER JOIN，导致结果少了 3000 行。验证框架 Level 3 数据集验证发现了这个错误。

铁律：LLM 只负责翻译特定的 Phoenix 特性片段，不碰 SQL 的其他部分。AST 引擎负责"组装"最终 SQL。

铁律 2：知识库 RAG 是 LLM 的"防幻觉护盾"

翻车案例：LLM 把 ARRAY_ELEM(arr, 1) 翻译成了达梦的 JSON_EXTRACT(arr, ‘$[0]’)，完全错误！因为达梦的数组访问语法是 arr[1]。

铁律：所有 Phoenix 函数→目标方言的映射，必须先查知识库。知识库没有的，LLM 生成的结果必须经过语法验证和数据集验证。

铁律 3：UPSERT 翻译必须考虑"全主键"边界情况

翻车案例：Phoenix 有一张日志表，所有列都是 PK（没有非主键列）。翻译成达梦 MERGE INTO 时，WHEN MATCHED THEN UPDATE SET 子句为空，达梦直接报错。

铁律：UPSERT 变异规则必须检测"非主键列是否为空"，空时使用 DO NOTHING（金仓）或自更新模式（达梦/OB）。

铁律 4：SALT_BUCKETS 的迁移必须咨询架构师

SALT_BUCKETS 的替代方案取决于查询模式：
点查为主 → Hash 分区表
范围扫描为主 → Range 分区表
混合负载 → Hash 分区 + 二级索引

铁律：SALT_BUCKETS 相关的 DDL 永远标记为 SEMI_AUTO 或 MANUAL，AI 生成方案后必须人工确认。

铁律 5：Phoenix Hint 一律移除或重新评估

Phoenix 的 /*+ RANGE_SCAN */ 是针对 HBase 的 Region 扫描优化，在关系库中完全没有意义。关系库有自己的查询优化器。

铁律：所有 Phoenix Hint 一律移除。如果查询在目标库上性能退化，由 DBA 根据目标库的 Hint 语法重新添加。

铁律 6：迁移后必须做"全量性能回归测试"

语义等价 ≠ 性能等价。Phoenix 的 RowKey 扫描在 HBase 上是 O(1)，翻译成关系库的索引查询可能是 O(log n)。

铁律：迁移后，用生产流量的镜像（Read Replica）做性能回归。P99 延迟超过原系统 2 倍的 SQL，标记为"需要索引优化"。

铁律 7：保留完整的迁移审计链

等保测评可能要求"证明每条 SQL 的迁移过程可追溯"。

铁律：每条 SQL 的迁移记录必须包含：
原始 SQL
特征指纹
迁移路径
AST 变换记录
AI 翻译 Prompt + Response
验证结果（四层）
最终翻译 SQL
审核人（如果有）

九、总结

老铁们，今天这篇硬核长文，把 Apache Phoenix → 国产数据库的 AI 辅助 SQL 迁移引擎，从 ANTLR4 方言解析器到 AST 变异规则引擎，从 LLM Agent 的 ReAct 模式到四层语义等价验证框架，从知识库 RAG 到批量编排引擎，连底裤都扒干净了。

龙虾开发者社区

小龙虾开发者社区是 CSDN 旗下专注 OpenClaw 生态的官方阵地，聚焦技能开发、插件实践与部署教程，为开发者提供可直接落地的方案、工具与交流平台，助力高效构建与落地 AI 应用

更多推荐

Agent Harness 架构真相：Prompt Cache 如何决定 Skill、MCP 与 SubAgent 设计

rompt Cache 这件事很容易被低估。Agent 每轮请求都可能带上系统提示词、工具定义、历史消息、文件片段、运行时提醒和工具结果。如果，缓存命中能明显降低延迟和成本；如果动态内容插得太靠前，从变化位置之后的内容就会失去缓存价值。Claude Code 很多看似绕的设计，本质上是在守这条边界。