TsDB2/migrate/mysql.go

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) + ")"
}