From a7395b08de53e38849ba5bfa973c7fa8e0ac55e6 Mon Sep 17 00:00:00 2001 From: hoteas <925970985@qq.com> Date: Fri, 20 Mar 2026 13:51:52 +0800 Subject: [PATCH] =?UTF-8?q?fix(cache):=20=E4=BF=AE=E5=A4=8D=20tableExists?= =?UTF-8?q?=20=E6=96=B9=E6=B3=95=E4=BB=A5=E6=AD=A3=E7=A1=AE=E6=A3=80?= =?UTF-8?q?=E6=B5=8B=E8=BE=BE=E6=A2=A6=E6=95=B0=E6=8D=AE=E5=BA=93=E4=B8=AD?= =?UTF-8?q?=E7=9A=84=E8=A1=A8=E5=AD=98=E5=9C=A8=E6=80=A7?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 修改 tableExists 方法，使用 COUNT(*) 查询当前 schema 下的表，避免因用户权限导致的误判 - 更新 ObjToFloat64、ObjToInt64 和 ObjToStr 方法，增加对多种整数和布尔类型的支持 - 在 Row 方法中处理 time.Time 类型，确保与 MySQL 兼容的字符串格式输出 --- cache/cache_db.go | 10 +- common/objtoobj.go | 55 ++++ db/query.go | 17 +- example/migrate/main.go | 599 ++++++++++++++++++++++++++++++++++++++++ 4 files changed, 673 insertions(+), 8 deletions(-) create mode 100644 example/migrate/main.go diff --git a/cache/cache_db.go b/cache/cache_db.go index 7050cb1..71ef592 100644 --- a/cache/cache_db.go +++ b/cache/cache_db.go @@ -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 diff --git a/common/objtoobj.go b/common/objtoobj.go index 87ea0b9..42b47f1 100644 --- a/common/objtoobj.go +++ b/common/objtoobj.go @@ -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 { diff --git a/db/query.go b/db/query.go index 9a5bdc6..d53c1ce 100644 --- a/db/query.go +++ b/db/query.go @@ -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 { diff --git a/example/migrate/main.go b/example/migrate/main.go new file mode 100644 index 0000000..a82b4ab --- /dev/null +++ b/example/migrate/main.go @@ -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（已完成）") +}