package migrate import ( "git.ycl.pub/yl32940/TsDB2/DB/core/page" "context" "database/sql" "fmt" "strings" "time" _ "github.com/go-sql-driver/mysql" // database/sql 驱动注册 ) // OpenMySQL 连接 MySQL。 // // dsn 就是 go-sql-driver/mysql 的 DSN, 例如: // // root:123456@tcp(127.0.0.1:3306)/vrtx_demo?charset=utf8mb4&parseTime=false // // 建议保持 parseTime=false (默认): 时间列按字符串原样搬。开了 parseTime 会引入 // 时区换算, 往返一次可能就变了几小时。 func OpenMySQL(dsn string, opts *Options) (*SQL, error) { db, err := sql.Open("mysql", dsn) if err != nil { return nil, fmt.Errorf("打开 MySQL: %w", err) } db.SetMaxOpenConns(4) if err := db.Ping(); err != nil { _ = db.Close() return nil, fmt.Errorf("连接 MySQL 失败: %w", err) } return newSQL(db, KindMySQL, dsn, mysqlDial{}, opts), nil } type mysqlDial struct{} func (mysqlDial) kind() Kind { return KindMySQL } // mysqlListTimeout 列库最多等这么久 —— 主机写错、端口不通时界面不该一直转圈。 const mysqlListTimeout = 10 * time.Second // MySQLDatabases 问一个 MySQL 服务器上有哪些库。 // // dsn 和 OpenMySQL 同样格式, 但库名可以留空: 能不能列出库只取决于账号权限, // 不取决于你已经知道哪个库 —— 连接界面正是靠它把库名做成候选, 免得手打错一个 // 字母换来一句"连接失败"。 // // 系统库 (mysql / information_schema 等) 不做过滤, 那是调用方的口味。 func MySQLDatabases(dsn string) ([]string, error) { db, err := sql.Open("mysql", dsn) if err != nil { return nil, fmt.Errorf("打开 MySQL: %w", err) } defer db.Close() db.SetMaxOpenConns(1) ctx, cancel := context.WithTimeout(context.Background(), mysqlListTimeout) defer cancel() rows, err := db.QueryContext(ctx, `SELECT SCHEMA_NAME FROM information_schema.SCHEMATA ORDER BY SCHEMA_NAME`) if err != nil { return nil, fmt.Errorf("列出数据库失败: %w", err) } defer rows.Close() var out []string for rows.Next() { var name string if err := rows.Scan(&name); err != nil { return nil, err } out = append(out, name) } if err := rows.Err(); err != nil { return nil, fmt.Errorf("列出数据库失败: %w", err) } return out, nil } func (mysqlDial) quote(name string) string { return "`" + strings.ReplaceAll(name, "`", "``") + "`" } func (d mysqlDial) columnDDL(c Column) string { return d.quote(c.Name) + " " + columnTypeName(c, true) + nullClause(c) } // createTable 建表。主键写成表级约束 PRIMARY KEY (col): VRTX 的行存储索引是单列, // 这里也只可能是单列。 func (d mysqlDial) createTable(s *TableSchema) string { parts := make([]string, 0, len(s.Columns)+1) for _, c := range s.Columns { parts = append(parts, " "+d.columnDDL(c)) } if s.PKColumn != "" { parts = append(parts, " PRIMARY KEY ("+d.quote(s.PKColumn)+")") } return fmt.Sprintf("CREATE TABLE %s (\n%s\n) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4", d.quote(s.Name), strings.Join(parts, ",\n")) } func (d mysqlDial) dropTable(table string) string { return "DROP TABLE IF EXISTS " + d.quote(table) } func (d mysqlDial) addColumn(table string, c Column) string { return fmt.Sprintf("ALTER TABLE %s ADD COLUMN %s", d.quote(table), d.columnDDL(c)) } func (d mysqlDial) createIndex(table string, idx IndexSchema) string { cols := make([]string, len(idx.Columns)) for i, c := range idx.Columns { cols[i] = d.quote(c) } return fmt.Sprintf("%s %s ON %s (%s)", createIndexKind(idx), d.quote(idx.Name), d.quote(table), strings.Join(cols, ", ")) } func (d mysqlDial) listTables(db *sql.DB) ([]string, error) { rows, err := db.Query(`SELECT TABLE_NAME FROM information_schema.TABLES WHERE TABLE_SCHEMA = DATABASE() AND TABLE_TYPE = 'BASE TABLE' ORDER BY TABLE_NAME`) if err != nil { return nil, fmt.Errorf("列 MySQL 表: %w", err) } defer rows.Close() var out []string for rows.Next() { var name string if err := rows.Scan(&name); err != nil { return nil, err } out = append(out, name) } return out, rows.Err() } // mysqlColumnDecl 把 information_schema 的长度信息收成 inferSQLType 认得的声明文本。 // // 字符列取【字节】上界, 不能取字符数: MySQL 的 varchar(50) 是 50 个【字符】, 在 // utf8mb4 下最多 200 字节; 而 VRTX 的定长档 (16/32/.../256 字节) 是按【字节】算容量的。 // 拿 50 去取档只会得到 64 字节 —— 一个 25 字的中文别名就装不下 (真实踩到: // resource_info.resAlias 是 varchar(50), 值 73 字节, 整表迁移失败)。 // CHARACTER_OCTET_LENGTH 正是"这列最多多少字节", 而且按该列自己的字符集算 // (latin1 仍是 1 字节/字符), 所以直接拿它当容量, 既够用又不至于无谓放大。 // // unsigned 必须单独传进来, 因为无符号这件事【不在 DATA_TYPE 里】: MySQL 对 // "bigint unsigned" 报的 DATA_TYPE 就是 "bigint", 只有 COLUMN_TYPE // ("bigint(20) unsigned") 保留 unsigned; 而 inferSQLType 正是靠声明原文里有没有 // "UNSIGNED" 来判无符号档的。少了这个后缀, 所有 MySQL 无符号列都会被当有符号收: // BIGINT UNSIGNED 的 2^63 读出来是负数, INT UNSIGNED 的上界只剩一半。 func mysqlColumnDecl(dataType string, unsigned bool, charBytes, numPrec, numScale int64) string { decl := dataType switch { case numScale > 0 || dataType == "decimal": // DECIMAL 必须带上 (p,s): 只给 "decimal" 会按 scale=0 收, 小数位被抹平。 decl = fmt.Sprintf("%s(%d,%d)", dataType, numPrec, numScale) case charBytes > 0: decl = fmt.Sprintf("%s(%d)", dataType, charBytes) } if unsigned { // 后缀放在最后: inferSQLType 用 strings.Contains 找 "UNSIGNED", 位置无所谓; // splitTypeArgs 只取第一个 "(" 里的参数, 后缀不会污染 (p,s) / (n)。 decl += " unsigned" } return decl } // mysqlUnsigned 判断 COLUMN_TYPE 是不是无符号列。 // // 只看 COLUMN_TYPE: DATA_TYPE 把 unsigned 丢了; EXTRA 只在显式写了 zerofill 时才 // 出现 "zerofill", 而 zerofill 本身也隐含 unsigned —— COLUMN_TYPE 两种都覆盖 // ("int(10) unsigned" / "int(10) unsigned zerofill")。 func mysqlUnsigned(columnType string) bool { return strings.Contains(strings.ToLower(columnType), "unsigned") } // mysqlBool 判断 COLUMN_TYPE 是不是 BOOLEAN 列。 // // 与 unsigned 同理, 这件事 DATA_TYPE 看不出来: MySQL 的 BOOLEAN 就是 tinyint(1), // information_schema 报的 DATA_TYPE 是 "tinyint"、COLUMN_TYPE 是 "tinyint(1)" —— // 只看 DATA_TYPE 会走 inferSQLType 的 TINYINT 分支收成 Int8, 而 sqltype.go 的 // "BOOLEAN"/"BOOL" 分支对 MySQL 源永远不可达。宽度不是 1 的 tinyint 是真·小整数, // 仍按 Int8 收; tinyint(1) unsigned 能装 0..255, 也不是真假语义, 不特判。 func mysqlBool(columnType string) bool { return strings.EqualFold(strings.TrimSpace(columnType), "tinyint(1)") } // columns 读列 + 单列主键名。 // // 主键的那点信息这里最全, 所以两件"VRTX 承载不了"的事在这里记下 (见 sourceNotes): // - 复合主键: VRTX 的行存储索引只支持单列。以前直接把主键名丢成空串, 下游只会报 // "源表没有主键"—— 而源表明明有主键, 只是多列, 提示是假的。 // - 列级的容量/精度差异: 只有这里看得到 information_schema 给的声明原文。 func (d mysqlDial) columns(db *sql.DB, table string) ([]Column, string, sourceNotes, error) { rows, err := db.Query(`SELECT COLUMN_NAME, DATA_TYPE, COLUMN_TYPE, COALESCE(CHARACTER_OCTET_LENGTH, 0), COALESCE(NUMERIC_PRECISION, 0), COALESCE(NUMERIC_SCALE, 0), COLUMN_KEY, IS_NULLABLE FROM information_schema.COLUMNS WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = ? ORDER BY ORDINAL_POSITION`, table) if err != nil { return nil, "", sourceNotes{}, err } defer rows.Close() var cols []Column var pkCols []string var notes sourceNotes for rows.Next() { // COLUMN_TYPE 只用来判无符号 (DATA_TYPE 把它丢了), 其余长度信息仍走各自那一列。 var name, dataType, columnType, key, nullable string var charBytes, numPrec, numScale int64 if err := rows.Scan(&name, &dataType, &columnType, &charBytes, &numPrec, &numScale, &key, &nullable); err != nil { return nil, "", sourceNotes{}, err } decl := mysqlColumnDecl(dataType, mysqlUnsigned(columnType), charBytes, numPrec, numScale) if mysqlBool(columnType) { // MySQL 的 BOOLEAN 就是 tinyint(1), DATA_TYPE 只报 "tinyint": 不特判只会 // 落到 Int8, 等值/索引给出的是整数语义而不是真假。换成 boolean 走 // inferSQLType 的 BOOL 分支, 先于常规 tinyint→Int8 路径生效。 decl = "boolean" } ft, param, note := inferSQLTypeWithNote(decl, true) if note != "" { notes.Capped = append(notes.Capped, name+": "+note) } if d := degradeNote(decl); d != "" { notes.Degraded = append(notes.Degraded, name+": "+d) } cols = append(cols, Column{ Name: name, Type: ft, Param: param, Decl: decl, TimeKind: sqlTimeKind(dataType), Nullable: nullable == "YES", // information_schema 的 IS_NULLABLE }) if key == "PRI" { pkCols = append(pkCols, name) } } if err := rows.Err(); err != nil { return nil, "", sourceNotes{}, err } if len(pkCols) == 1 { return cols, pkCols[0], notes, nil } if len(pkCols) > 1 { notes.PK = compositePKNote(table, pkCols) } // 复合主键 / 无主键: 都当作"没有单列主键", 由调用方决定补 _id 还是报错。 return cols, "", notes, nil } // indexes 读二级索引, 唯一性取 NON_UNIQUE=0。 // // STATISTICS 里同一个索引每列一行, NON_UNIQUE 各行相同, 用 MAX 收成一行; // 主键索引 (INDEX_NAME='PRIMARY') 由 PKColumn 表达, 排除在外。 func (d mysqlDial) indexes(db *sql.DB, table string) ([]IndexSchema, error) { rows, err := db.Query(`SELECT INDEX_NAME, MAX(NON_UNIQUE), GROUP_CONCAT(COLUMN_NAME ORDER BY SEQ_IN_INDEX) FROM information_schema.STATISTICS WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = ? AND INDEX_NAME <> 'PRIMARY' GROUP BY INDEX_NAME`, table) if err != nil { return nil, err } defer rows.Close() var out []IndexSchema for rows.Next() { var name, cols string var nonUnique int if err := rows.Scan(&name, &nonUnique, &cols); err != nil { return nil, err } out = append(out, IndexSchema{ Name: name, // SQL 侧分不出 Hash / BTree: 默认按 B+ 树建; 真正的类型由 __vrtx_meta // 带回来 (见 applyMeta) Type: page.IndexTypeBTree, Columns: strings.Split(cols, ","), Unique: nonUnique == 0, }) } return out, rows.Err() } func (d mysqlDial) metaDDL() string { return fmt.Sprintf(`CREATE TABLE IF NOT EXISTS %s ( tbl VARCHAR(191) NOT NULL, payload LONGTEXT NOT NULL, PRIMARY KEY (tbl) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4`, d.quote(MetaTable)) } func (d mysqlDial) metaUpsert() string { return fmt.Sprintf("INSERT INTO %s (tbl, payload) VALUES (?, ?) "+ "ON DUPLICATE KEY UPDATE payload = VALUES(payload)", d.quote(MetaTable)) } func (d mysqlDial) metaSelect() string { return fmt.Sprintf("SELECT payload FROM %s WHERE tbl = ? LIMIT 1", d.quote(MetaTable)) } // maxPlaceholders MySQL 单语句占位符上限是 65535, 留一点余量。 func (d mysqlDial) maxPlaceholders() int { return 60000 } func (d mysqlDial) textLenExpr(col string) string { // MySQL 的 LENGTH() 返回【字节】数 (CHAR_LENGTH 才是字符数), 与按字节计容量的 // 宽度码同口径。 return "LENGTH(" + d.quote(col) + ")" }