hoteas
a7395b08de
- 修改 tableExists 方法，使用 COUNT(*) 查询当前 schema 下的表，避免因用户权限导致的误判 - 更新 ObjToFloat64、ObjToInt64 和 ObjToStr 方法，增加对多种整数和布尔类型的支持 - 在 Row 方法中处理 time.Time 类型，确保与 MySQL 兼容的字符串格式输出
600 lines
17 KiB
Go
600 lines
17 KiB
Go
// 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（已完成）")
|
||
}
|