fix(cache): 修复 tableExists 方法以正确检测达梦数据库中的表存在性
- 修改 tableExists 方法，使用 COUNT(*) 查询当前 schema 下的表，避免因用户权限导致的误判 - 更新 ObjToFloat64、ObjToInt64 和 ObjToStr 方法，增加对多种整数和布尔类型的支持 - 在 Row 方法中处理 time.Time 类型，确保与 MySQL 兼容的字符串格式输出
This commit is contained in:
Vendored
+7
-3
@@ -136,9 +136,13 @@ func (that *CacheDb) tableExists(tableName string) bool {
|
||||
return len(res) != 0
|
||||
|
||||
case "dm", "dameng":
|
||||
// 双引号创建的表名保留原始大小写，不能 ToUpper
|
||||
res := that.Db.Query(`SELECT TABLE_NAME FROM USER_TABLES WHERE TABLE_NAME='` + tableName + `'`)
|
||||
return len(res) != 0
|
||||
// USER_TABLES 仅含当前登录用户（如 SYSDBA）拥有的表。
|
||||
// 当连接 schema（如 ZWBG）与登录用户不同时，ZWBG 的表不会出现在
|
||||
// USER_TABLES 中，导致 tableExists 误判为"不存在"，进而触发重复建表失败。
|
||||
// 改为直接 SELECT COUNT(*) 探测当前 schema 下的可访问性：
|
||||
// 表存在则返回 1 行，表不存在则报错返回空结果（nil）。
|
||||
res := that.Db.Query(`SELECT COUNT(*) as cnt FROM "` + tableName + `"`)
|
||||
return len(res) > 0
|
||||
}
|
||||
|
||||
return false
|
||||
|
||||
@@ -214,8 +214,28 @@ func ObjToFloat64(obj interface{}, e ...*Error) float64 {
|
||||
switch val := obj.(type) {
|
||||
case int:
|
||||
v = float64(val)
|
||||
case int8:
|
||||
v = float64(val)
|
||||
case int16:
|
||||
v = float64(val)
|
||||
case int32:
|
||||
v = float64(val)
|
||||
case int64:
|
||||
v = float64(val)
|
||||
case uint:
|
||||
v = float64(val)
|
||||
case uint8:
|
||||
v = float64(val)
|
||||
case uint16:
|
||||
v = float64(val)
|
||||
case uint32:
|
||||
v = float64(val)
|
||||
case uint64:
|
||||
v = float64(val)
|
||||
case bool:
|
||||
if val {
|
||||
v = 1
|
||||
}
|
||||
case string:
|
||||
value, e2 := strconv.ParseFloat(val, 64)
|
||||
if e2 != nil {
|
||||
@@ -295,8 +315,28 @@ func ObjToInt64(obj interface{}, e ...*Error) int64 {
|
||||
switch val := obj.(type) {
|
||||
case int:
|
||||
v = int64(val)
|
||||
case int8:
|
||||
v = int64(val)
|
||||
case int16:
|
||||
v = int64(val)
|
||||
case int32:
|
||||
v = int64(val)
|
||||
case int64:
|
||||
v = val
|
||||
case uint:
|
||||
v = int64(val)
|
||||
case uint8:
|
||||
v = int64(val)
|
||||
case uint16:
|
||||
v = int64(val)
|
||||
case uint32:
|
||||
v = int64(val)
|
||||
case uint64:
|
||||
v = int64(val)
|
||||
case bool:
|
||||
if val {
|
||||
v = 1
|
||||
}
|
||||
case string:
|
||||
value, e2 := StrToInt(val)
|
||||
if e2 != nil {
|
||||
@@ -375,8 +415,20 @@ func ObjToStr(obj interface{}) string {
|
||||
switch val := obj.(type) {
|
||||
case int:
|
||||
str = strconv.Itoa(val)
|
||||
case int8:
|
||||
str = strconv.Itoa(int(val))
|
||||
case int16:
|
||||
str = strconv.Itoa(int(val))
|
||||
case int32:
|
||||
str = strconv.Itoa(int(val))
|
||||
case uint8:
|
||||
str = strconv.Itoa(int(val))
|
||||
case uint16:
|
||||
str = strconv.Itoa(int(val))
|
||||
case uint32:
|
||||
str = strconv.FormatUint(uint64(val), 10)
|
||||
case uint64:
|
||||
str = strconv.FormatUint(val, 10)
|
||||
case int64:
|
||||
str = strconv.FormatInt(val, 10)
|
||||
case []byte:
|
||||
@@ -389,6 +441,9 @@ func ObjToStr(obj interface{}) string {
|
||||
str = strconv.FormatFloat(float64(val), 'f', -1, 32)
|
||||
case bool:
|
||||
str = strconv.FormatBool(val)
|
||||
case time.Time:
|
||||
// DM8等数据库驱动直接返回time.Time，格式化为MySQL兼容的字符串
|
||||
str = val.Format("2006-01-02 15:04:05")
|
||||
default:
|
||||
strbte, err := json.MarshalIndent(obj, "", "\t")
|
||||
if err == nil {
|
||||
|
||||
+12
-5
@@ -8,6 +8,7 @@ import (
|
||||
"reflect"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
. "code.hoteas.com/golang/hotime/common"
|
||||
)
|
||||
@@ -546,22 +547,28 @@ func (that *HoTimeDB) Row(resl *sql.Rows) []Map {
|
||||
return nil
|
||||
}
|
||||
for j := 0; j < colCount; j++ {
|
||||
// 统一转小写，兼容DM8等返回大写列名的数据库驱动
|
||||
colKey := strings.ToLower(strs[j])
|
||||
val := a[j]
|
||||
if val == nil {
|
||||
lis[strs[j]] = nil
|
||||
lis[colKey] = nil
|
||||
continue
|
||||
}
|
||||
switch v := val.(type) {
|
||||
case []byte:
|
||||
lis[strs[j]] = convertBytes(v, categories[j], scales[j])
|
||||
lis[colKey] = convertBytes(v, categories[j], scales[j])
|
||||
case float64:
|
||||
lis[strs[j]] = fixFloatValue(v, categories[j], scales[j])
|
||||
lis[colKey] = fixFloatValue(v, categories[j], scales[j])
|
||||
case float32:
|
||||
lis[strs[j]] = fixFloatValue(float64(v), categories[j], scales[j])
|
||||
lis[colKey] = fixFloatValue(float64(v), categories[j], scales[j])
|
||||
case time.Time:
|
||||
// DM8驱动直接返回time.Time，格式化为MySQL兼容字符串
|
||||
lis[colKey] = v.Format("2006-01-02 15:04:05")
|
||||
default:
|
||||
lis[strs[j]] = val
|
||||
lis[colKey] = val
|
||||
}
|
||||
}
|
||||
|
||||
dest = append(dest, lis)
|
||||
}
|
||||
if err := resl.Err(); err != nil {
|
||||
|
||||
@@ -0,0 +1,599 @@
|
||||
// migrate: MySQL 8.0 → 达梦 DM8 直连迁移工具
|
||||
// 用法：在 hotimev1.5 目录下执行
|
||||
// go run example/migrate/main.go --config d:/work/myst-go/config/config.json
|
||||
package main
|
||||
|
||||
import (
|
||||
"database/sql"
|
||||
"encoding/json"
|
||||
"flag"
|
||||
"fmt"
|
||||
"log"
|
||||
"os"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
_ "gitee.com/chunanyong/dm"
|
||||
_ "github.com/go-sql-driver/mysql"
|
||||
)
|
||||
|
||||
// ─────────────────── 配置读取 ───────────────────
|
||||
|
||||
type DBConf struct {
|
||||
Host string `json:"host"`
|
||||
Port string `json:"port"`
|
||||
Name string `json:"name"`
|
||||
User string `json:"user"`
|
||||
Password string `json:"password"`
|
||||
Prefix string `json:"prefix"`
|
||||
}
|
||||
|
||||
type Config struct {
|
||||
DB map[string]DBConf `json:"db"`
|
||||
}
|
||||
|
||||
func loadConfig(path string) (*Config, error) {
|
||||
data, err := os.ReadFile(path)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var raw map[string]json.RawMessage
|
||||
if err = json.Unmarshal(data, &raw); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var dbRaw map[string]DBConf
|
||||
if err = json.Unmarshal(raw["db"], &dbRaw); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &Config{DB: dbRaw}, nil
|
||||
}
|
||||
|
||||
func getMySQLConf(cfg *Config) (DBConf, error) {
|
||||
// 1. 精确匹配 "mysql"（启用态）
|
||||
if v, ok := cfg.DB["mysql"]; ok {
|
||||
return v, nil
|
||||
}
|
||||
// 2. 迁移后改为 "mysql-" 的情况（原始主源）
|
||||
if v, ok := cfg.DB["mysql-"]; ok {
|
||||
log.Printf("[注意] MySQL 配置 key=mysql- 已禁用，仍用于迁移源（host=%s db=%s）", v.Host, v.Name)
|
||||
return v, nil
|
||||
}
|
||||
// 3. 其他以 mysql 开头且不以 _ 结尾的 key（排除 mysql_ 这类备用库）
|
||||
for k, v := range cfg.DB {
|
||||
if strings.HasPrefix(k, "mysql") && !strings.HasSuffix(k, "_") {
|
||||
log.Printf("[注意] 使用 MySQL 配置 key=%q (host=%s db=%s)", k, v.Host, v.Name)
|
||||
return v, nil
|
||||
}
|
||||
}
|
||||
return DBConf{}, fmt.Errorf("config.json 中未找到 mysql 配置（期望 key=mysql 或 mysql-）")
|
||||
}
|
||||
|
||||
func getDMConf(cfg *Config) (DBConf, error) {
|
||||
for k, v := range cfg.DB {
|
||||
if strings.HasPrefix(k, "dm") && !strings.HasSuffix(k, "-") {
|
||||
return v, nil
|
||||
}
|
||||
}
|
||||
for k, v := range cfg.DB {
|
||||
if strings.HasPrefix(k, "dm") {
|
||||
return v, nil
|
||||
}
|
||||
}
|
||||
return DBConf{}, fmt.Errorf("config.json 中未找到 dm 配置")
|
||||
}
|
||||
|
||||
// ─────────────────── 类型转换 ───────────────────
|
||||
|
||||
func mysqlTypeToDM(dataType string, charLen *int64, numPrec, numScale *int64) string {
|
||||
dt := strings.ToLower(dataType)
|
||||
switch dt {
|
||||
case "int", "integer", "mediumint", "smallint", "tinyint", "year", "bit":
|
||||
return "INT"
|
||||
case "bigint":
|
||||
return "BIGINT"
|
||||
case "float":
|
||||
return "FLOAT"
|
||||
case "double", "double precision", "real":
|
||||
return "DOUBLE"
|
||||
case "decimal", "numeric":
|
||||
p := int64(10)
|
||||
s := int64(0)
|
||||
if numPrec != nil {
|
||||
p = *numPrec
|
||||
}
|
||||
if numScale != nil {
|
||||
s = *numScale
|
||||
}
|
||||
return fmt.Sprintf("DECIMAL(%d,%d)", p, s)
|
||||
case "varchar", "nvarchar":
|
||||
l := int64(255)
|
||||
if charLen != nil {
|
||||
l = *charLen
|
||||
}
|
||||
if l > 32767 {
|
||||
return "CLOB"
|
||||
}
|
||||
return fmt.Sprintf("VARCHAR(%d)", l)
|
||||
case "char", "nchar":
|
||||
l := int64(1)
|
||||
if charLen != nil {
|
||||
l = *charLen
|
||||
}
|
||||
return fmt.Sprintf("CHAR(%d)", l)
|
||||
case "datetime", "timestamp":
|
||||
return "TIMESTAMP"
|
||||
case "date":
|
||||
return "DATE"
|
||||
case "time":
|
||||
return "TIME"
|
||||
case "text", "tinytext", "mediumtext", "longtext":
|
||||
return "CLOB"
|
||||
case "blob", "tinyblob", "mediumblob", "longblob", "binary", "varbinary":
|
||||
return "BLOB"
|
||||
case "json":
|
||||
return "CLOB"
|
||||
case "enum", "set":
|
||||
return "VARCHAR(100)"
|
||||
default:
|
||||
return "VARCHAR(255)"
|
||||
}
|
||||
}
|
||||
|
||||
// q 用双引号包裹标识符（小写）
|
||||
func q(name string) string {
|
||||
return `"` + strings.ToLower(name) + `"`
|
||||
}
|
||||
|
||||
// ─────────────────── 列信息 ───────────────────
|
||||
|
||||
type Column struct {
|
||||
Name string
|
||||
DataType string
|
||||
CharLen *int64
|
||||
NumPrec *int64
|
||||
NumScale *int64
|
||||
IsNullable bool
|
||||
Default *string
|
||||
Extra string
|
||||
Comment string
|
||||
ColKey string
|
||||
}
|
||||
|
||||
type Index struct {
|
||||
Name string
|
||||
IsUnique bool
|
||||
ColNames []string
|
||||
}
|
||||
|
||||
func getColumns(mysqlDB *sql.DB, dbName, tableName string) ([]Column, error) {
|
||||
rows, err := mysqlDB.Query(`
|
||||
SELECT COLUMN_NAME, DATA_TYPE, CHARACTER_MAXIMUM_LENGTH,
|
||||
NUMERIC_PRECISION, NUMERIC_SCALE,
|
||||
IS_NULLABLE, COLUMN_DEFAULT, EXTRA, COLUMN_COMMENT, COLUMN_KEY
|
||||
FROM INFORMATION_SCHEMA.COLUMNS
|
||||
WHERE TABLE_SCHEMA=? AND TABLE_NAME=?
|
||||
ORDER BY ORDINAL_POSITION`, dbName, tableName)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
var cols []Column
|
||||
for rows.Next() {
|
||||
var c Column
|
||||
var nullable string
|
||||
err = rows.Scan(&c.Name, &c.DataType, &c.CharLen, &c.NumPrec, &c.NumScale,
|
||||
&nullable, &c.Default, &c.Extra, &c.Comment, &c.ColKey)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
c.IsNullable = (nullable == "YES")
|
||||
cols = append(cols, c)
|
||||
}
|
||||
return cols, rows.Err()
|
||||
}
|
||||
|
||||
func getIndexes(mysqlDB *sql.DB, dbName, tableName string) ([]Index, error) {
|
||||
rows, err := mysqlDB.Query(`
|
||||
SELECT INDEX_NAME, NON_UNIQUE, COLUMN_NAME, SEQ_IN_INDEX
|
||||
FROM INFORMATION_SCHEMA.STATISTICS
|
||||
WHERE TABLE_SCHEMA=? AND TABLE_NAME=? AND INDEX_NAME!='PRIMARY'
|
||||
ORDER BY INDEX_NAME, SEQ_IN_INDEX`, dbName, tableName)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
indexMap := map[string]*Index{}
|
||||
var order []string
|
||||
for rows.Next() {
|
||||
var idxName, colName string
|
||||
var nonUnique, seqInIndex int
|
||||
if err = rows.Scan(&idxName, &nonUnique, &colName, &seqInIndex); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if _, ok := indexMap[idxName]; !ok {
|
||||
indexMap[idxName] = &Index{Name: idxName, IsUnique: nonUnique == 0}
|
||||
order = append(order, idxName)
|
||||
}
|
||||
indexMap[idxName].ColNames = append(indexMap[idxName].ColNames, colName)
|
||||
}
|
||||
var result []Index
|
||||
for _, name := range order {
|
||||
result = append(result, *indexMap[name])
|
||||
}
|
||||
return result, rows.Err()
|
||||
}
|
||||
|
||||
// ─────────────────── DDL 生成 ───────────────────
|
||||
|
||||
func buildCreateTable(tableName string, cols []Column) string {
|
||||
var lines []string
|
||||
var pkCol string
|
||||
|
||||
for _, c := range cols {
|
||||
dmType := mysqlTypeToDM(c.DataType, c.CharLen, c.NumPrec, c.NumScale)
|
||||
isAuto := strings.Contains(strings.ToLower(c.Extra), "auto_increment")
|
||||
isPK := c.ColKey == "PRI"
|
||||
|
||||
var parts []string
|
||||
parts = append(parts, fmt.Sprintf(" %s %s", q(c.Name), dmType))
|
||||
|
||||
if isPK && isAuto {
|
||||
parts = append(parts, "IDENTITY(1,1) NOT NULL")
|
||||
pkCol = c.Name
|
||||
} else {
|
||||
if !c.IsNullable {
|
||||
parts = append(parts, "NOT NULL")
|
||||
}
|
||||
if c.Default != nil {
|
||||
dv := *c.Default
|
||||
upper := strings.ToUpper(dv)
|
||||
switch {
|
||||
case upper == "CURRENT_TIMESTAMP" || upper == "NOW()":
|
||||
parts = append(parts, "DEFAULT CURRENT_TIMESTAMP")
|
||||
case dmType == "INT" || dmType == "BIGINT" || dmType == "FLOAT" ||
|
||||
dmType == "DOUBLE" || strings.HasPrefix(dmType, "DECIMAL"):
|
||||
parts = append(parts, "DEFAULT "+dv)
|
||||
case dmType == "CLOB" || dmType == "BLOB":
|
||||
// 不设默认值
|
||||
default:
|
||||
safe := strings.ReplaceAll(dv, "'", "''")
|
||||
parts = append(parts, fmt.Sprintf("DEFAULT '%s'", safe))
|
||||
}
|
||||
}
|
||||
}
|
||||
lines = append(lines, strings.Join(parts, " "))
|
||||
}
|
||||
|
||||
// 主键约束
|
||||
if pkCol != "" {
|
||||
lines = append(lines, fmt.Sprintf(" PRIMARY KEY (%s)", q(pkCol)))
|
||||
} else {
|
||||
var pkCols []string
|
||||
for _, c := range cols {
|
||||
if c.ColKey == "PRI" {
|
||||
pkCols = append(pkCols, q(c.Name))
|
||||
}
|
||||
}
|
||||
if len(pkCols) > 0 {
|
||||
lines = append(lines, fmt.Sprintf(" PRIMARY KEY (%s)", strings.Join(pkCols, ", ")))
|
||||
}
|
||||
}
|
||||
|
||||
return fmt.Sprintf("CREATE TABLE %s (\n%s\n)", q(tableName), strings.Join(lines, ",\n"))
|
||||
}
|
||||
|
||||
// ─────────────────── DM8 表操作 ───────────────────
|
||||
|
||||
func tableExistsInDM(dmDB *sql.DB, tableName string) bool {
|
||||
var cnt int
|
||||
err := dmDB.QueryRow(fmt.Sprintf(`SELECT COUNT(*) FROM %s WHERE ROWNUM=1`, q(tableName))).Scan(&cnt)
|
||||
return err == nil
|
||||
}
|
||||
|
||||
func createTableInDM(dmDB *sql.DB, tableName string, cols []Column, indexes []Index,
|
||||
tableComment string) error {
|
||||
|
||||
// 如果已存在则 DROP
|
||||
if tableExistsInDM(dmDB, tableName) {
|
||||
log.Printf(" 表 %s 已存在，先删除 ...", tableName)
|
||||
_, err := dmDB.Exec(fmt.Sprintf(`DROP TABLE %s CASCADE`, q(tableName)))
|
||||
if err != nil {
|
||||
return fmt.Errorf("DROP TABLE %s: %w", tableName, err)
|
||||
}
|
||||
}
|
||||
|
||||
ddl := buildCreateTable(tableName, cols)
|
||||
log.Printf(" CREATE TABLE %s ...", tableName)
|
||||
if _, err := dmDB.Exec(ddl); err != nil {
|
||||
return fmt.Errorf("CREATE TABLE %s: %w\nDDL:\n%s", tableName, err, ddl)
|
||||
}
|
||||
|
||||
// 表注释
|
||||
if tableComment != "" {
|
||||
safe := strings.ReplaceAll(tableComment, "'", "''")
|
||||
stmt := fmt.Sprintf("COMMENT ON TABLE %s IS '%s'", q(tableName), safe)
|
||||
if _, err := dmDB.Exec(stmt); err != nil {
|
||||
log.Printf(" [警告] 表注释失败: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// 列注释
|
||||
for _, c := range cols {
|
||||
if c.Comment != "" {
|
||||
safe := strings.ReplaceAll(c.Comment, "'", "''")
|
||||
stmt := fmt.Sprintf("COMMENT ON COLUMN %s.%s IS '%s'", q(tableName), q(c.Name), safe)
|
||||
if _, err := dmDB.Exec(stmt); err != nil {
|
||||
log.Printf(" [警告] 列注释 %s.%s 失败: %v", tableName, c.Name, err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// 索引
|
||||
for _, idx := range indexes {
|
||||
idxName := tableName + "_" + idx.Name
|
||||
uniqueKW := ""
|
||||
if idx.IsUnique {
|
||||
uniqueKW = "UNIQUE "
|
||||
}
|
||||
colList := make([]string, len(idx.ColNames))
|
||||
for i, cn := range idx.ColNames {
|
||||
colList[i] = q(cn)
|
||||
}
|
||||
stmt := fmt.Sprintf("CREATE %sINDEX %s ON %s (%s)",
|
||||
uniqueKW, q(idxName), q(tableName), strings.Join(colList, ", "))
|
||||
if _, err := dmDB.Exec(stmt); err != nil {
|
||||
log.Printf(" [警告] 创建索引 %s 失败: %v", idxName, err)
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// ─────────────────── 数据迁移 ───────────────────
|
||||
|
||||
// hasIdentityCol 判断列表中是否有自增主键列
|
||||
func hasIdentityCol(cols []Column) bool {
|
||||
for _, c := range cols {
|
||||
if strings.Contains(strings.ToLower(c.Extra), "auto_increment") && c.ColKey == "PRI" {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
func migrateData(mysqlDB, dmDB *sql.DB, tableName string, cols []Column, batchSize int) (int64, error) {
|
||||
colNames := make([]string, len(cols))
|
||||
dmColNames := make([]string, len(cols))
|
||||
placeholders := make([]string, len(cols))
|
||||
for i, c := range cols {
|
||||
colNames[i] = "`" + c.Name + "`"
|
||||
dmColNames[i] = q(c.Name)
|
||||
placeholders[i] = "?"
|
||||
}
|
||||
|
||||
insertSQL := fmt.Sprintf("INSERT INTO %s (%s) VALUES (%s)",
|
||||
q(tableName),
|
||||
strings.Join(dmColNames, ", "),
|
||||
strings.Join(placeholders, ", "))
|
||||
|
||||
// 统计总行数
|
||||
var total int64
|
||||
_ = mysqlDB.QueryRow(fmt.Sprintf("SELECT COUNT(*) FROM `%s`", tableName)).Scan(&total)
|
||||
|
||||
// DM8：有 IDENTITY 列时，需要 SET IDENTITY_INSERT ON 才能插入指定 ID
|
||||
if hasIdentityCol(cols) {
|
||||
if _, err := dmDB.Exec(fmt.Sprintf("SET IDENTITY_INSERT %s ON", q(tableName))); err != nil {
|
||||
log.Printf(" [警告] SET IDENTITY_INSERT ON 失败（将继续）: %v", err)
|
||||
}
|
||||
defer func() {
|
||||
_, _ = dmDB.Exec(fmt.Sprintf("SET IDENTITY_INSERT %s OFF", q(tableName)))
|
||||
}()
|
||||
}
|
||||
|
||||
stmt, err := dmDB.Prepare(insertSQL)
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("Prepare INSERT %s: %w", tableName, err)
|
||||
}
|
||||
defer stmt.Close()
|
||||
|
||||
var offset int64
|
||||
var migrated int64
|
||||
for {
|
||||
selectSQL := fmt.Sprintf("SELECT %s FROM `%s` LIMIT %d OFFSET %d",
|
||||
strings.Join(colNames, ", "), tableName, batchSize, offset)
|
||||
rows, err := mysqlDB.Query(selectSQL)
|
||||
if err != nil {
|
||||
return migrated, err
|
||||
}
|
||||
|
||||
vals := make([]interface{}, len(cols))
|
||||
valPtrs := make([]interface{}, len(cols))
|
||||
for i := range vals {
|
||||
valPtrs[i] = &vals[i]
|
||||
}
|
||||
|
||||
rowCount := 0
|
||||
for rows.Next() {
|
||||
if err = rows.Scan(valPtrs...); err != nil {
|
||||
rows.Close()
|
||||
return migrated, err
|
||||
}
|
||||
// 处理 []byte → string（DM8 CLOB 等）
|
||||
args := make([]interface{}, len(cols))
|
||||
for i, v := range vals {
|
||||
if b, ok := v.([]byte); ok {
|
||||
args[i] = string(b)
|
||||
} else {
|
||||
args[i] = v
|
||||
}
|
||||
}
|
||||
if _, err = stmt.Exec(args...); err != nil {
|
||||
rows.Close()
|
||||
return migrated, fmt.Errorf("INSERT row into %s: %w", tableName, err)
|
||||
}
|
||||
migrated++
|
||||
rowCount++
|
||||
}
|
||||
rows.Close()
|
||||
if err = rows.Err(); err != nil {
|
||||
return migrated, err
|
||||
}
|
||||
if rowCount == 0 {
|
||||
break
|
||||
}
|
||||
offset += int64(batchSize)
|
||||
|
||||
if total > 0 {
|
||||
pct := float64(migrated) * 100 / float64(total)
|
||||
fmt.Printf("\r 进度: %d/%d (%.1f%%)", migrated, total, pct)
|
||||
}
|
||||
}
|
||||
if total > 0 {
|
||||
fmt.Println()
|
||||
}
|
||||
return migrated, nil
|
||||
}
|
||||
|
||||
// ─────────────────── 主程序 ───────────────────
|
||||
|
||||
var skipTables = map[string]bool{
|
||||
"hotime_cache": true,
|
||||
"cached": true,
|
||||
}
|
||||
|
||||
func main() {
|
||||
configPath := flag.String("config", "d:/work/myst-go/config/config.json", "配置文件路径")
|
||||
batchSize := flag.Int("batch", 500, "每批迁移行数")
|
||||
onlyTables := flag.String("only", "", "只迁移指定表（逗号分隔）")
|
||||
flag.Parse()
|
||||
|
||||
log.SetFlags(log.LstdFlags | log.Lmsgprefix)
|
||||
log.SetPrefix("[migrate] ")
|
||||
|
||||
// 读配置
|
||||
log.Printf("读取配置: %s", *configPath)
|
||||
cfg, err := loadConfig(*configPath)
|
||||
if err != nil {
|
||||
log.Fatalf("读取配置失败: %v", err)
|
||||
}
|
||||
|
||||
mysqlConf, err := getMySQLConf(cfg)
|
||||
if err != nil {
|
||||
log.Fatalf("%v", err)
|
||||
}
|
||||
dmConf, err := getDMConf(cfg)
|
||||
if err != nil {
|
||||
log.Fatalf("%v", err)
|
||||
}
|
||||
|
||||
log.Printf("源库 MySQL : %s:%s db=%s user=%s", mysqlConf.Host, mysqlConf.Port, mysqlConf.Name, mysqlConf.User)
|
||||
log.Printf("目标 DM8 : %s:%s schema=%s user=%s", dmConf.Host, dmConf.Port, dmConf.Name, dmConf.User)
|
||||
|
||||
// 连接 MySQL
|
||||
mysqlDSN := fmt.Sprintf("%s:%s@tcp(%s:%s)/%s?charset=utf8mb4&parseTime=True&loc=Local",
|
||||
mysqlConf.User, mysqlConf.Password, mysqlConf.Host, mysqlConf.Port, mysqlConf.Name)
|
||||
log.Printf("连接 MySQL ...")
|
||||
mysqlDB, err := sql.Open("mysql", mysqlDSN)
|
||||
if err != nil {
|
||||
log.Fatalf("sql.Open mysql: %v", err)
|
||||
}
|
||||
defer mysqlDB.Close()
|
||||
if err = mysqlDB.Ping(); err != nil {
|
||||
log.Fatalf("MySQL Ping 失败: %v\nDSN=%s", err, mysqlDSN)
|
||||
}
|
||||
log.Printf("MySQL 连接成功")
|
||||
|
||||
// 连接 DM8
|
||||
// DSN 格式: dm://user:password@host:port?schema=SCHEMA
|
||||
dmDSN := fmt.Sprintf("dm://%s:%s@%s:%s?schema=%s",
|
||||
dmConf.User, dmConf.Password, dmConf.Host, dmConf.Port, dmConf.Name)
|
||||
log.Printf("连接 DM8 ...")
|
||||
dmDB, err := sql.Open("dm", dmDSN)
|
||||
if err != nil {
|
||||
log.Fatalf("sql.Open dm: %v", err)
|
||||
}
|
||||
defer dmDB.Close()
|
||||
dmDB.SetConnMaxLifetime(5 * time.Minute)
|
||||
dmDB.SetMaxOpenConns(3)
|
||||
if err = dmDB.Ping(); err != nil {
|
||||
log.Fatalf("DM8 Ping 失败: %v\nDSN=%s", err, dmDSN)
|
||||
}
|
||||
log.Printf("DM8 连接成功")
|
||||
|
||||
// 枚举 MySQL 表
|
||||
tables, err := mysqlDB.Query(`
|
||||
SELECT TABLE_NAME, IFNULL(TABLE_COMMENT,'')
|
||||
FROM INFORMATION_SCHEMA.TABLES
|
||||
WHERE TABLE_SCHEMA=? AND TABLE_TYPE='BASE TABLE'
|
||||
ORDER BY TABLE_NAME`, mysqlConf.Name)
|
||||
if err != nil {
|
||||
log.Fatalf("枚举表失败: %v", err)
|
||||
}
|
||||
type tableInfo struct{ name, comment string }
|
||||
var allTables []tableInfo
|
||||
for tables.Next() {
|
||||
var ti tableInfo
|
||||
tables.Scan(&ti.name, &ti.comment)
|
||||
allTables = append(allTables, ti)
|
||||
}
|
||||
tables.Close()
|
||||
|
||||
// 过滤
|
||||
onlySet := map[string]bool{}
|
||||
if *onlyTables != "" {
|
||||
for _, t := range strings.Split(*onlyTables, ",") {
|
||||
onlySet[strings.TrimSpace(t)] = true
|
||||
}
|
||||
}
|
||||
var toMigrate []tableInfo
|
||||
for _, ti := range allTables {
|
||||
if skipTables[ti.name] {
|
||||
log.Printf("跳过（框架内置）: %s", ti.name)
|
||||
continue
|
||||
}
|
||||
if len(onlySet) > 0 && !onlySet[ti.name] {
|
||||
log.Printf("跳过（未指定）: %s", ti.name)
|
||||
continue
|
||||
}
|
||||
toMigrate = append(toMigrate, ti)
|
||||
}
|
||||
|
||||
log.Printf("共 %d 张表待迁移", len(toMigrate))
|
||||
fmt.Println(strings.Repeat("─", 60))
|
||||
|
||||
// 逐表迁移
|
||||
var totalRows int64
|
||||
start := time.Now()
|
||||
for i, ti := range toMigrate {
|
||||
log.Printf("[%d/%d] 迁移表: %s", i+1, len(toMigrate), ti.name)
|
||||
|
||||
cols, err := getColumns(mysqlDB, mysqlConf.Name, ti.name)
|
||||
if err != nil {
|
||||
log.Fatalf(" 读取列失败 %s: %v", ti.name, err)
|
||||
}
|
||||
indexes, err := getIndexes(mysqlDB, mysqlConf.Name, ti.name)
|
||||
if err != nil {
|
||||
log.Printf(" [警告] 读取索引失败 %s: %v", ti.name, err)
|
||||
}
|
||||
|
||||
// 建表
|
||||
if err = createTableInDM(dmDB, ti.name, cols, indexes, ti.comment); err != nil {
|
||||
log.Fatalf(" 建表失败 %s: %v", ti.name, err)
|
||||
}
|
||||
|
||||
// 迁移数据
|
||||
migrated, err := migrateData(mysqlDB, dmDB, ti.name, cols, *batchSize)
|
||||
if err != nil {
|
||||
log.Fatalf(" 数据迁移失败 %s: %v", ti.name, err)
|
||||
}
|
||||
totalRows += migrated
|
||||
log.Printf(" ✓ %s 完成，迁移 %d 行", ti.name, migrated)
|
||||
}
|
||||
|
||||
elapsed := time.Since(start)
|
||||
fmt.Println(strings.Repeat("─", 60))
|
||||
log.Printf("迁移完成！共 %d 表，%d 行，耗时 %s", len(toMigrate), totalRows, elapsed.Round(time.Millisecond))
|
||||
log.Printf("下一步：确认数据无误后，config.json 中 mysql→mysql-，dm-→dm（已完成）")
|
||||
}
|
||||
Reference in New Issue
Block a user