# 跨数据库SQL兼容性检测方案
## 一、检测目标
检测系统中所有SQL语句在目标数据库（达梦/MySQL/Oracle/DB2）上的兼容性，识别并修复不兼容的SQL语法，确保系统平滑迁移到新数据库环境。
---
## 二、检测范围
### 检测对象
1. **Mapper XML文件**：所有MyBatis Mapper XML中的SQL语句
2. **Java代码**：使用注解（@Select/@Insert/@Update/@Delete）的SQL
3. **动态SQL**：Java代码中拼接的SQL字符串
4. **数据库脚本**：DDL/DML脚本文件
### 检测项
1. **SQL语法兼容性**
- MySQL特有语法 → 达梦/Oracle语法
- 分页、函数、操作符等
2. **数据类型兼容性**
- MySQL特有类型 → 标准SQL类型
- 类型映射关系
3. **字符串和日期处理**
- 字符串拼接、日期函数等
---
## 三、MySQL到达梦兼容性对照表
| SQL类型 | MySQL语法 | 达梦语法 | 兼容性说明 | 检测方法 |
|--------|-----------|---------|-----------|---------|
| **分页** | `LIMIT 10, 20` | `LIMIT 20 OFFSET 10` 或 `ROWNUM` | 达梦8+支持LIMIT | Grep搜索`LIMIT` |
| **自增主键** | `AUTO_INCREMENT` | `IDENTITY` 或 `SEQUENCE` | 需修改为IDENTITY | Grep搜索`AUTO_INCREMENT` |
| **日期函数** | `NOW()` | `SYSDATE` | 需替换为SYSDATE | Grep搜索`NOW()` |
| **字符串拼接** | `CONCAT(a,b)` 或 `CONCAT_WS()` | `a \|\| b` 或 `CONCAT()` | 达梦支持两种方式 | Grep搜索`CONCAT` |
| **反引号** | `` `table` `` | `"table"` | 需替换为双引号 | Grep搜索反引号 |
| **布尔值** | `TRUE/FALSE` | `1/0` | 需替换为1/0 | Grep搜索`TRUE\|FALSE` |
| **注释符号** | `#注释` | `-- 注释` | `#`不支持，需改为`--` | Grep搜索`^[[:space:]]*#` |
| **当前时间** | `CURRENT_TIMESTAMP` | `SYSDATE` | 建议统一用SYSDATE | Grep搜索`CURRENT_TIMESTAMP` |
| **空字符串** | `''` | `''` 或 `NULL` | 大部分兼容 | 人工检查 |
| **数据类型** | `TINYINT/MEDIUMINT` | `TINYINT/INT` | TINYINT需调整为INT | Grep搜索`TINYINT\|MEDIUMINT` |
| **索引提示** | `USE INDEX` | 不支持 | 需删除索引提示 | Grep搜索`USE INDEX` |
| **GROUP BY** | 允许SELECT非聚合字段 | 要求SELECT字段都在GROUP BY中 | 需修改SQL | 人工检查 |
| **IF函数** | `IF(condition, a, b)` | `CASE WHEN` | 需改写为CASE WHEN | Grep搜索`\bIF\(` |
| **日期格式化** | `DATE_FORMAT()` | `TO_CHAR()` | 需改写为TO_CHAR | Grep搜索`DATE_FORMAT` |
| **字符串转数字** | `CAST('123' AS SIGNED)` | `CAST('123' AS INT)` | SIGNED改为INT | Grep搜索`AS SIGNED` |
---
## 四、检测方法
### 方案A：命令行快速扫描
```bash
#!/bin/bash
# MySQL到达梦SQL兼容性扫描脚本
echo "开始扫描MySQL特有SQL语法..."
echo ""
# 1. 搜索分页语句（LIMIT）
echo "=== 1. 分页语句 (LIMIT) ==="
grep -rn "LIMIT\s" src/main/resources/mybatis/
echo ""
# 2. 搜索自增主键（AUTO_INCREMENT）
echo "=== 2. 自增主键 (AUTO_INCREMENT) ==="
grep -rn "AUTO_INCREMENT" src/
echo ""
# 3. 搜索日期函数（NOW）
echo "=== 3. 日期函数 (NOW) ==="
grep -rn "NOW()" src/
echo ""
# 4. 搜索反引号（MySQL特有）
echo "=== 4. 反引号 (MySQL特有) ==="
grep -rn '`' src/main/resources/mybatis/
echo ""
# 5. 搜索布尔值（TRUE/FALSE）
echo "=== 5. 布尔值 (TRUE/FALSE) ==="
grep -rn "\bTRUE\b\|\bFALSE\b" src/
echo ""
# 6. 搜索注释符号（#）
echo "=== 6. 注释符号 (#) ==="
grep -rn "^[[:space:]]*#" src/main/resources/mybatis/
echo ""
# 7. 搜索特定数据类型
echo "=== 7. MySQL特有数据类型 (TINYINT/MEDIUMINT) ==="
grep -rn "TINYINT\|MEDIUMINT" src/
echo ""
# 8. 搜索IF函数
echo "=== 8. IF函数 (需改写为CASE WHEN) ==="
grep -rn "\bIF(" src/
echo ""
# 9. 搜索DATE_FORMAT
echo "=== 9. DATE_FORMAT (需改写为TO_CHAR) ==="
grep -rn "DATE_FORMAT" src/
echo ""
# 10. 搜索CAST AS SIGNED
echo "=== 10. CAST AS SIGNED (需改为AS INT) ==="
grep -rn "AS SIGNED" src/
echo ""
# 11. 搜索USE INDEX
echo "=== 11. USE INDEX (达梦不支持) ==="
grep -rn "USE INDEX\|FORCE INDEX\|IGNORE INDEX" src/
echo ""
# 12. 搜索GROUP_CONCAT
echo "=== 12. GROUP_CONCAT (需改写为LISTAGG) ==="
grep -rn "GROUP_CONCAT" src/
echo ""
echo "扫描完成！"
```
**Windows版本（PowerShell）**：
```powershell
# MySQL到达梦SQL兼容性扫描脚本
Write-Host "开始扫描MySQL特有SQL语法..." -ForegroundColor Green
Write-Host ""
# 定义要扫描的模式
$patterns = @{
"分页语句 (LIMIT)" = "LIMIT\s"
"自增主键 (AUTO_INCREMENT)" = "AUTO_INCREMENT"
"日期函数 (NOW)" = "NOW\(\)"
"反引号" = "`""
"布尔值 (TRUE/FALSE)" = "\bTRUE\b|\bFALSE\b"
"注释符号 (#)" = "^\s*#"
"TINYINT/MEDIUMINT" = "TINYINT|MEDIUMINT"
"IF函数" = "\bIF\("
"DATE_FORMAT" = "DATE_FORMAT"
"CAST AS SIGNED" = "AS SIGNED"
"USE INDEX" = "USE INDEX|FORCE INDEX|IGNORE INDEX"
"GROUP_CONCAT" = "GROUP_CONCAT"
}
# 执行扫描
foreach ($item in $patterns.GetEnumerator()) {
Write-Host "=== $($item.Key) ===" -ForegroundColor Yellow
Select-String -Path "src\**\*.xml", "src\**\*.java" -Pattern $item.Value -Recurse
Write-Host ""
}
Write-Host "扫描完成！" -ForegroundColor Green
```
---
### 方案B：Python自动化检测工具
```python
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
MySQL到达梦SQL兼容性检测工具
"""
import re
import json
from pathlib import Path
from typing import Dict, List, Tuple
from dataclasses import dataclass, asdict
from datetime import datetime
@dataclass
class CompatibilityIssue:
"""兼容性问题"""
issue_type: str
severity: str # HIGH, MEDIUM, LOW
file_path: str
line_number: int
mysql_pattern: str
dm_pattern: str
description: str
suggestion: str
class MySQLToDMCompatibilityChecker:
"""MySQL到达梦兼容性检测器"""
# 检测规则配置
RULES = {
'LIMIT': {
'pattern': r'\bLIMIT\s+(\d+)\s*,\s*(\d+)',
'severity': 'HIGH',
'mysql_pattern': 'LIMIT offset, count',
'dm_pattern': 'LIMIT count OFFSET offset',
'description': 'MySQL分页语法不兼容达梦',
'suggestion': '将 LIMIT offset, count 改为 LIMIT count OFFSET offset'
},
'AUTO_INCREMENT': {
'pattern': r'\bAUTO_INCREMENT\b',
'severity': 'HIGH',
'mysql_pattern': 'AUTO_INCREMENT',
'dm_pattern': 'IDENTITY',
'description': '达梦不支持AUTO_INCREMENT',
'suggestion': '将 AUTO_INCREMENT 改为 IDENTITY 或使用序列SEQUENCE'
},
'NOW': {
'pattern': r'\bNOW\(\)',
'severity': 'HIGH',
'mysql_pattern': 'NOW()',
'dm_pattern': 'SYSDATE',
'description': '达梦不支持NOW()函数',
'suggestion': '将 NOW() 改为 SYSDATE'
},
'BACKTICK': {
'pattern': r'`',
'severity': 'HIGH',
'mysql_pattern': '`table_name`',
'dm_pattern': '"table_name"',
'description': 'MySQL反引号语法',
'suggestion': '将反引号改为双引号 "table_name"'
},
'BOOLEAN': {
'pattern': r'\b(TRUE|FALSE)\b',
'severity': 'MEDIUM',
'mysql_pattern': 'TRUE/FALSE',
'dm_pattern': '1/0',
'description': '达梦不支持布尔值关键字',
'suggestion': '将 TRUE 改为 1，FALSE 改为 0'
},
'COMMENT_HASH': {
'pattern': r'^\s*#',
'severity': 'LOW',
'mysql_pattern': '# 注释',
'dm_pattern': '-- 注释',
'description': '达梦不支持#注释',
'suggestion': '将 # 注释 改为 -- 注释'
},
'TINYINT': {
'pattern': r'\bTINYINT\b',
'severity': 'MEDIUM',
'mysql_pattern': 'TINYINT',
'dm_pattern': 'INT或SMALLINT',
'description': '达梦不完全支持TINYINT',
'suggestion': '将 TINYINT 改为 SMALLINT 或 INT'
},
'IF_FUNCTION': {
'pattern': r'\bIF\s*\(',
'severity': 'HIGH',
'mysql_pattern': 'IF(condition, a, b)',
'dm_pattern': 'CASE WHEN condition THEN a ELSE b END',
'description': '达梦不支持IF()函数',
'suggestion': '将 IF(condition, a, b) 改为 CASE WHEN condition THEN a ELSE b END'
},
'DATE_FORMAT': {
'pattern': r'\bDATE_FORMAT\s*\(',
'severity': 'HIGH',
'mysql_pattern': 'DATE_FORMAT(date, format)',
'dm_pattern': 'TO_CHAR(date, format)',
'description': '达梦不支持DATE_FORMAT函数',
'suggestion': '将 DATE_FORMAT(date, format) 改为 TO_CHAR(date, format)'
},
'CAST_SIGNED': {
'pattern': r'\bCAST\s*\([^)]+\s+AS\s+SIGNED\b',
'severity': 'MEDIUM',
'mysql_pattern': 'CAST(expr AS SIGNED)',
'dm_pattern': 'CAST(expr AS INT)',
'description': '达梦不支持SIGNED类型',
'suggestion': '将 CAST(expr AS SIGNED) 改为 CAST(expr AS INT)'
},
'USE_INDEX': {
'pattern': r'\b(USE|FORCE|IGNORE)\s+INDEX\b',
'severity': 'MEDIUM',
'mysql_pattern': 'USE INDEX (index_name)',
'dm_pattern': '(删除索引提示)',
'description': '达梦不支持索引提示',
'suggestion': '删除 USE INDEX/FORCE INDEX/IGNORE INDEX'
},
'GROUP_CONCAT': {
'pattern': r'\bGROUP_CONCAT\s*\(',
'severity': 'HIGH',
'mysql_pattern': 'GROUP_CONCAT(expr)',
'dm_pattern': 'LISTAGG(expr, delimiter)',
'description': '达梦不支持GROUP_CONCAT函数',
'suggestion': '将 GROUP_CONCAT(expr) 改为 LISTAGG(expr, ",")'
}
}
def __init__(self, source_dir: str):
"""
初始化检测器
:param source_dir: 源代码目录
"""
self.source_dir = Path(source_dir)
self.issues: List[CompatibilityIssue] = []
def check_file(self, file_path: Path) -> List[CompatibilityIssue]:
"""检测单个文件"""
issues = []
content = file_path.read_text(encoding='utf-8')
lines = content.split('\n')
for line_no, line in enumerate(lines, 1):
for rule_name, rule in self.RULES.items():
matches = re.finditer(rule['pattern'], line, re.IGNORECASE)
for match in matches:
issues.append(CompatibilityIssue(
issue_type=rule_name,
severity=rule['severity'],
file_path=str(file_path.relative_to(self.source_dir)),
line_number=line_no,
mysql_pattern=rule['mysql_pattern'],
dm_pattern=rule['dm_pattern'],
description=rule['description'],
suggestion=rule['suggestion']
))
return issues
def check_all(self) -> List[CompatibilityIssue]:
"""检测所有文件"""
# 检测XML文件
xml_files = list(self.source_dir.rglob('**/*.xml'))
# 检测Java文件
java_files = list(self.source_dir.rglob('**/*.java'))
all_files = xml_files + java_files
print(f'开始检测 {len(all_files)} 个文件...\n')
for idx, file_path in enumerate(all_files, 1):
print(f'[{idx}/{len(all_files)}] 检测: {file_path.name}')
issues = self.check_file(file_path)
self.issues.extend(issues)
return self.issues
def generate_report(self, output_file: str = 'compatibility_report.json'):
"""生成检测报告"""
report = {
'scan_time': datetime.now().isoformat(),
'summary': {
'total_issues': len(self.issues),
'by_severity': self._count_by_severity(),
'by_type': self._count_by_type(),
'by_file': self._count_by_file()
},
'issues': [asdict(issue) for issue in self.issues]
}
# 保存JSON报告
with open(output_file, 'w', encoding='utf-8') as f:
json.dump(report, f, ensure_ascii=False, indent=2)
# 生成Markdown报告
md_file = output_file.replace('.json', '.md')
self._generate_markdown_report(report, md_file)
print(f'\n检测完成！')
print(f'JSON报告: {output_file}')
print(f'Markdown报告: {md_file}')
print(f'\n总计发现 {report["summary"]["total_issues"]} 个兼容性问题：')
for severity, count in report['summary']['by_severity'].items():
print(f' - {severity}: {count}')
def _count_by_severity(self) -> Dict[str, int]:
"""按严重程度统计"""
count = {'HIGH': 0, 'MEDIUM': 0, 'LOW': 0}
for issue in self.issues:
count[issue.severity] += 1
return count
def _count_by_type(self) -> Dict[str, int]:
"""按问题类型统计"""
count = {}
for issue in self.issues:
if issue.issue_type not in count:
count[issue.issue_type] = 0
count[issue.issue_type] += 1
return dict(sorted(count.items(), key=lambda x: x[1], reverse=True))
def _count_by_file(self) -> Dict[str, int]:
"""按文件统计"""
count = {}
for issue in self.issues:
if issue.file_path not in count:
count[issue.file_path] = 0
count[issue.file_path] += 1
return dict(sorted(count.items(), key=lambda x: x[1], reverse=True)[:10])
def _generate_markdown_report(self, report: dict, output_file: str):
"""生成Markdown格式报告"""
with open(output_file, 'w', encoding='utf-8') as f:
f.write('# MySQL到达梦SQL兼容性检测报告\n\n')
f.write(f'**扫描时间**: {report["scan_time"]}\n\n')
# 摘要
f.write('## 检测摘要\n\n')
f.write(f'- **总问题数**: {report["summary"]["total_issues"]}\n')
f.write('\n### 按严重程度\n\n')
f.write('| 严重程度 | 数量 |\n')
f.write('|---------|------|\n')
for severity, count in report['summary']['by_severity'].items():
f.write(f'| {severity} | {count} |\n')
f.write('\n### 按问题类型\n\n')
f.write('| 问题类型 | 数量 |\n')
f.write('|---------|------|\n')
for issue_type, count in list(report['summary']['by_type'].items())[:10]:
f.write(f'| {issue_type} | {count} |\n')
f.write('\n### 问题最多的文件 (Top 10)\n\n')
f.write('| 文件 | 问题数 |\n')
f.write('|------|--------|\n')
for file_path, count in report['summary']['by_file'].items():
f.write(f'| {file_path} | {count} |\n')
# 详细问题列表
f.write('\n## 详细问题清单\n\n')
for issue in report['issues']:
f.write(f'### {issue["issue_type"]} - {issue["severity"]}\n\n')
f.write(f'**文件**: `{issue["file_path"]}:{issue["line_number"]}`\n\n')
f.write(f'**问题描述**: {issue["description"]}\n\n')
f.write(f'- **MySQL语法**: `{issue["mysql_pattern"]}`\n')
f.write(f'- **达梦语法**: `{issue["dm_pattern"]}`\n')
f.write(f'- **修复建议**: {issue["suggestion"]}\n\n')
f.write('---\n\n')
def generate_fix_script(self, output_file: str = 'fix_compatibility.sh'):
"""生成自动修复脚本（仅适用于简单替换）"""
with open(output_file, 'w', encoding='utf-8') as f:
f.write('#!/bin/bash\n')
f.write('# MySQL到达梦SQL兼容性自动修复脚本\n')
f.write('# 注意：此脚本仅处理简单替换，复杂情况需人工确认\n\n')
# 可自动修复的规则
fixable_rules = {
'NOW': ('NOW()', 'SYSDATE'),
'AUTO_INCREMENT': ('AUTO_INCREMENT', 'IDENTITY'),
'BOOLEAN_TRUE': ('\\bTRUE\\b', '1'),
'BOOLEAN_FALSE': ('\\bFALSE\\b', '0'),
}
for rule_name, (old, new) in fixable_rules.items():
f.write(f'# 修复 {rule_name}\n')
f.write(f'find src -type f -name "*.xml" -o -name "*.java" | \\\n')
f.write(f' xargs sed -i "s/{old}/{new}/g"\n\n')
f.write('echo "自动修复完成！请仔细检查修改内容。"\n')
print(f'自动修复脚本已生成: {output_file}')
print('⚠️ 注意：使用前请先提交代码，修复脚本可能产生副作用')
def main():
"""主函数"""
import sys
# 配置
SOURCE_DIR = 'src'
JSON_REPORT = 'check/reports/compatibility_report.json'
FIX_SCRIPT = 'check/scripts/fix_compatibility.sh'
# 创建检测器
checker = MySQLToDMCompatibilityChecker(SOURCE_DIR)
# 执行检测
checker.check_all()
# 生成报告
checker.generate_report(JSON_REPORT)
# 生成修复脚本
checker.generate_fix_script(FIX_SCRIPT)
if __name__ == '__main__':
main()
```
---
## 五、常见SQL兼容性问题及修复示例
### 1. 分页查询
**MySQL写法**：
```xml
```
**达梦写法**：
```xml
```
**或者使用ROWNUM**（达梦7及以下）：
```xml
```
---
### 2. 日期函数
**MySQL写法**：
```xml
```
**达梦写法**：
```xml
```
---
### 3. IF函数改为CASE WHEN
**MySQL写法**：
```xml
```
**达梦写法**：
```xml
```
---
### 4. GROUP_CONCAT改为LISTAGG
**MySQL写法**：
```xml
```
**达梦写法**：
```xml
```
---
### 5. 字符串拼接
**MySQL写法**：
```xml
```
**达梦写法**（两种都支持）：
```xml
```
或
```xml
```
---
### 6. CAST类型转换
**MySQL写法**：
```xml
```
**达梦写法**：
```xml
```
---
### 7. DATE_FORMAT改为TO_CHAR
**MySQL写法**：
```xml
```
**达梦写法**：
```xml
```
---
## 六、检测步骤
### 1. 准备阶段
```bash
# 创建报告目录
mkdir -p check/reports
mkdir -p check/scripts
# 统计需要检测的文件数量
find src -name "*.xml" | wc -l
find src -name "*.java" | wc -l
```
### 2. 执行检测
**方式1：使用命令行快速扫描**
```bash
# Linux/Mac
bash check/scripts/scan_mysql_syntax.sh > scan_results.txt
# Windows PowerShell
powershell -ExecutionPolicy Bypass -File check/scripts/scan_mysql_syntax.ps1 > scan_results.txt
```
**方式2：使用Python自动化检测**
```bash
# 运行检测脚本
python check/scripts/mysql_to_dm_checker.py
# 查看报告
cat check/reports/compatibility_report.json
cat check/reports/compatibility_report.md
```
### 3. 分析报告
根据报告中的问题优先级处理：
1. **HIGH（高危）**
- 影响SQL执行的关键语法
- 必须修复才能正常运行
2. **MEDIUM（中危）**
- 可能有兼容性问题
- 建议修复
3. **LOW（低危）**
- 次要兼容性问题
- 可选修复
### 4. 修复问题
**自动修复**（谨慎使用）：
```bash
# 先提交代码
git add .
git commit -m "备份：修复前的代码"
# 执行自动修复脚本
bash check/scripts/fix_compatibility.sh
# 检查修复结果
git diff
```
**手动修复**：
根据报告中的文件路径和行号，逐一修复兼容性问题。
### 5. 验证修复
```bash
# 重新扫描确认问题已修复
python check/scripts/mysql_to_dm_checker.py
# 对比修复前后的报告
diff check/reports/before_fix.json check/reports/after_fix.json
```
---
## 七、输出报告
### JSON格式报告（兼容性检测机器报告）
```json
{
"scan_time": "2025-01-05T14:30:00",
"summary": {
"total_issues": 45,
"by_severity": {
"HIGH": 23,
"MEDIUM": 15,
"LOW": 7
},
"by_type": {
"NOW": 18,
"LIMIT": 12,
"IF_FUNCTION": 8,
"AUTO_INCREMENT": 5,
"BOOLEAN": 2
},
"by_file": {
"src/main/resources/mybatis/mapper/aiccs/abnormal/AbnormalListMapper.xml": 8,
"src/main/java/com/chinaweal/aiccs/aiccs/seriousillegal/mapper/SeriousIllegalMapper.java": 5
}
},
"issues": [
{
"issue_type": "NOW",
"severity": "HIGH",
"file_path": "src/main/resources/mybatis/mapper/aiccs/abnormal/AbnormalListMapper.xml",
"line_number": 45,
"mysql_pattern": "NOW()",
"dm_pattern": "SYSDATE",
"description": "达梦不支持NOW()函数",
"suggestion": "将 NOW() 改为 SYSDATE"
}
]
}
```
### Markdown格式报告（兼容性检测人工报告）
参见工具生成的Markdown格式报告。
---
## 八、注意事项
1. **数据库版本差异**
- 达梦8与达梦7语法有差异
- 确认目标数据库版本
2. **索引提示**
- 达梦不支持MySQL的索引提示语法
- 删除后可能影响查询性能
3. **GROUP BY严格模式**
- 达梦要求SELECT中的非聚合字段必须在GROUP BY中
- MySQL的宽松模式在达梦中会报错
4. **字符串拼接**
- 达梦支持 `||` 和 `CONCAT()` 两种方式
- 建议统一使用 `CONCAT()` 以提高代码可读性
5. **NULL处理**
- 达梦的NULL处理与MySQL可能有细微差异
- 注意空字符串 `''` 和 `NULL` 的区别
6. **自动生成的SQL**
- MyBatis-Plus等框架生成的SQL也需要检测
- 可能需要配置框架的数据库方言
---
## 九、相关文件
- Mapper XML路径: `src/main/resources/mybatis/mapper/{数据源}/**/*.xml`
- 扫描脚本: `check/scripts/scan_mysql_syntax.sh` / `scan_mysql_syntax.ps1`
- 检测工具: `check/scripts/mysql_to_dm_checker.py`
- 检测报告: `check/reports/compatibility_report.json` / `compatibility_report.md`
- 修复脚本: `check/scripts/fix_compatibility.sh`