TsDB2/migrate/run.go

package migrate

import (
	"git.ycl.pub/yl32940/TsDB2/DB/core"
	"git.ycl.pub/yl32940/TsDB2/DB/core/page"
	"git.ycl.pub/yl32940/TsDB2/DB/core/types"
	"errors"
	"fmt"
	"strings"
	"time"
)

// errStop 内部哨兵: 回调拿它请求"提前结束扫描"(Limit 生效), 由 Run 吞掉,
// 不会当成失败往外抛。
var errStop = errors.New("migrate: 提前结束扫描")

// Committer 目标端可选实现: 每张表写完调用一次, 让引擎把这一批落盘。
// VRTX 实现了它 (千万行的库不能等最后 Close 才提交)。
type Committer interface {
	Commit() error
}

// Run 把 src 的表结构 + 数据 + 索引 + 主键搬到 dst。
//
//	VRTX → SQLite / MySQL    导出
//	SQLite / MySQL → VRTX    导入
//	VRTX → VRTX              整库拷贝 / 压实
//
// 任意方向共用这一条路径: src.Schema 给出中间表示, src.Scan 推流, dst.CreateTable
// 建结构 (含索引/主键), dst.Open 拿批量写入器。opts 传 nil 等价于 DefaultOptions()。
//
// 出错时返回的部分报告仍可用 (已完成的表照常记着), 方便断点续迁 —— 重跑时带
// 同样的 opts 但 Recreate=false, 已迁完的表会走"复用已有表 + 追加", 会重复插数据,
// 所以务实用 Tables 只点名没迁完的表。
//
// 开关 ContinueOnError 决定"一张表出问题要不要连坐整趟": 打开时失败的表记进
// 报告的 TableReport.Error 并继续迁后面的表, 最后 Run 用聚合错误返回失败清单;
// 关闭时(默认)第一个错误就中止 —— 中止的那张表同样带着 Error, 只是不再往下走。
func Run(src Source, dst Sink, opts *Options) (*Report, error) {
	if src == nil || dst == nil {
		return nil, errors.New("migrate: 源与目标都需要提供")
	}
	o := opts.normalized()
	if src.Identity() == dst.Identity() {
		return nil, fmt.Errorf("migrate: 源与目标是同一个库 (%s), 拒绝自我迁移", src.Identity())
	}

	started := time.Now()
	rep := &Report{From: src.Kind(), To: dst.Kind()}
	defer func() { rep.Elapsed = time.Since(started) }()

	plan, skipped, err := planTables(src, o)
	if err != nil {
		return rep, err
	}
	rep.Tables = append(rep.Tables, skipped...)
	if len(plan) == 0 && len(skipped) == 0 {
		return rep, nil
	}

	var failed []error
	for i, name := range plan {
		// 先报"开始": 表一多、或头一张就是大表时, 只有"完成"行的日志会让人
		// 以为卡住了, 分不清是在跑还是死锁。
		o.vlogf("[migrate] 表 %s (%d/%d): 开始", name, i+1, len(plan))
		tr, err := runTable(src, dst, name, o)
		if err != nil {
			// 原因先落进报告再说 —— 跳过还是中止都是后话, 但"为什么失败"丢了就
			// 只剩一句"失败", 界面和报告都无从解释。
			tr.Error = err.Error()
		}
		if err != nil && o.ContinueOnError {
			// 单表失败不拖垮整趟: 记进报告接着迁后面的表, 全部跑完再统一报错。
			failed = append(failed, err)
			o.logf("[migrate] 表 %s 失败, 跳过继续: %v", name, err)
		}
		rep.Tables = append(rep.Tables, tr)
		rep.Rows += tr.Rows
		if err != nil && !o.ContinueOnError {
			return rep, err
		}
	}
	if len(failed) > 0 {
		return rep, fmt.Errorf("有 %d 张表没能迁完 (其余已照常迁完): %w", len(failed), errors.Join(failed...))
	}
	return rep, nil
}

// planTables 决定迁哪些表。返回 (待迁清单, 直接标记为跳过的报告)。
func planTables(src Source, o *Options) ([]string, []TableReport, error) {
	names, err := src.Tables()
	if err != nil {
		return nil, nil, fmt.Errorf("列出源库的表: %w", err)
	}
	avail := make(map[string]bool, len(names))
	for _, n := range names {
		avail[n] = true
	}

	var (
		plan    []string
		skipped []TableReport
	)
	if len(o.Tables) == 0 {
		for _, n := range names {
			// 保真表是迁移的"随身行李", 不是业务表: 不带 __vrtx_meta 去 VRTX,
			// 除非调用方点名要它。
			if n == MetaTable {
				continue
			}
			plan = append(plan, n)
		}
		return plan, nil, nil
	}
	for _, want := range o.Tables {
		if !avail[want] {
			skipped = append(skipped, TableReport{Table: want, Skipped: true, Note: "源库没有这张表"})
			continue
		}
		plan = append(plan, want)
	}
	return plan, skipped, nil
}

// progressEvery 进行中进度的上报间隔: 太密会把日志刷没, 太疏又看不出"在动"。
const progressEvery = 2 * time.Second

// abortWriter 出错时收掉写入器。
//
// 走 Discarder 而不是 Close: Close 是【成功语义】(SQL 侧 flush 剩余缓冲 + Commit),
// 用它收尾等于把缓冲区里那批行当成功提交。扫描失败 / flush 失败的调用方都带着错误
// 返回, 库里却躺了半张表 —— 这种"报了错但还是写进去了"最难查。
// 没实现 Discarder 的写入器 (VRTX) 退回 Close: 它的 Close 不提交, 提交由 Committer
// 在整表成功后单独调用, 因此没有这个问题。
func abortWriter(w RowWriter) {
	if d, ok := w.(Discarder); ok {
		_ = d.Discard()
		return
	}
	_ = w.Close()
}

// runTable 迁一张表: 读结构 → (必要时) 建表 → 推流写入 → 提交。
// 第二个返回值是这张表的报告 (即使出错也带着已写行数, 便于定位)。
func runTable(src Source, dst Sink, name string, o *Options) (TableReport, error) {
	begin := time.Now()
	tr := TableReport{Table: name}

	sc, err := src.Schema(name)
	if err != nil {
		return tr, fmt.Errorf("读表 %s 的结构: %w", name, err)
	}
	sc = sc.Clone() // 迁移会往 schema 里补列, 不能污染源端那份

	// TEXT 采样定宽 (见 textwidth.go): 建目标表之前把"源声明 TEXT 家族"的动态列按
	// 源端采样的实际最大长度改落成定长档。先于 prepareForVRTX / fitRowWidthForVRTX:
	// 退化主键列要不要补索引、行宽闸门要不要把采样得过宽的列降回动态列, 都该看到
	// 采样后的最终列宽。DryRun 不碰数据, 连采样也不做。
	textWidthNote := ""
	if dst.Kind() == KindVRTX && o.TextWidthScan && !o.DryRun {
		textWidthNote = scanTextWidths(src, sc, o)
	}

	if dst.Kind() == KindVRTX {
		if err := prepareForVRTX(sc, o); err != nil {
			return tr, err
		}
		// 结构闸门必须在建表【之前】: 源库的结构能力比 VRTX 宽 (MySQL 1017 列 / 表名 64 字节 /
		// 复合主键…), 不先查一遍就会在建表时才炸, 报错还看不出是"源库比目标库宽"
		// (见 validateForVRTX)。放在 prepareForVRTX 之后是因为补出的 _id 列也算列数。
		if err := validateForVRTX(sc); err != nil {
			return tr, err
		}
		// 行宽闸门 (同样必须在建表之前, 且要在列清单定形之后): 源库那张表可能宽到 VRTX
		// 一页装不下两行 —— 那是建出来之后【每一行】都写不进去的死局, 不能等写数据才发现。
		// 能自动救的就地救 (宽度 > 256 字节的定长字符串列 → 动态列), 救不了才拒绝,
		// 见 fitRowWidthForVRTX 的文件头。
		demoted, err := fitRowWidthForVRTX(sc)
		if err != nil {
			return tr, err
		}
		if demoted > 0 {
			o.vlogf("[migrate] 表 %s 行宽超限: 已把 %d 个宽列降为动态列", sc.Name, demoted)
		}
	}
	tr.Columns = len(sc.Columns)
	tr.Indexes = len(sc.Indexes)
	if sc.PKColumn != "" {
		tr.Indexes++ // 主键索引也是一条索引
	}

	if o.DryRun {
		tr.Note = "试跑"
		o.vlogf("[migrate] %s → %s 表 %s: %d 列 %d 索引",
			src.Kind(), dst.Kind(), name, tr.Columns, tr.Indexes)
		tr.Elapsed = time.Since(begin)
		return tr, nil
	}

	exists, err := dst.Exists(name)
	if err != nil {
		return tr, fmt.Errorf("检查目标表 %s: %w", name, err)
	}
	if !exists && !o.Create {
		return tr, fmt.Errorf("目标库没有表 %s, 且未开启建表 (Options.Create)", name)
	}
	if exists && o.Recreate {
		o.vlogf("[migrate] 目标表 %s 已存在, 按 Recreate 重建", name)
	}
	if err := dst.CreateTable(sc, o.Recreate); err != nil {
		return tr, err
	}
	if len(sc.SkippedIndexes) > 0 {
		// 跳过的索引必须报出来: 静默丢索引等于悄悄改了结构。
		tr.Note = appendNote(tr.Note, "跳过索引 "+strings.Join(sc.SkippedIndexes, "; "))
	}
	if sc.PKNote != "" {
		// 源表的主键没能原样承载 (复合主键)。这是"结构变了", 比丢一条索引更要紧,
		// 单独报一句而不是塞进别的列表里。
		tr.Note = appendNote(tr.Note, sc.PKNote)
	}
	if len(sc.DegradedColumns) > 0 {
		// 类型从定点退成浮点、主键降级成普通列这类: 类型都换了, 存进去的东西语义不同。
		// 与容量封顶分开报 —— 那个是"上限变小, 类型没变"。
		tr.Note = appendNote(tr.Note, "降级 "+strings.Join(sc.DegradedColumns, "; "))
	}
	if len(sc.CappedColumns) > 0 {
		// 同理: 源库声明了 60000 字节的列到这里只能装 types.MaxStringSize, 不报等于
		// 悄悄把容量改小。超限只影响"这一列的上限比声明的小", 数据照迁 (真有大值会在
		// 写入时报出列名), 所以是报告而不是拒绝整张表。
		tr.Note = appendNote(tr.Note, "容量封顶 "+strings.Join(sc.CappedColumns, "; "))
	}
	if textWidthNote != "" {
		// TEXT 采样定宽的结果 (哪些列定到了哪一档 / 哪些列超宽保持动态): 定宽会约束
		// 未来写入, 必须让用户在报告里看得见。
		tr.Note = appendNote(tr.Note, textWidthNote)
	}
	if o.SchemaOnly {
		// 两阶段迁移的第一步: 到这儿结构就算立好了, 数据一行不碰。
		tr.Note = appendNote(tr.Note, "仅结构")
		tr.Elapsed = time.Since(begin)
		o.vlogf("[migrate] %s 表 %s: 结构就绪 (%d 列 %d 索引)",
			dst.Kind(), name, tr.Columns, tr.Indexes)
		return tr, nil
	}

	writer, err := dst.Open(name, sc)
	if err != nil {
		return tr, err
	}

	var (
		written uint64
		batch   = make([]Row, 0, o.Batch)
		lastRep = time.Now()
	)
	flush := func() error {
		if len(batch) == 0 {
			return nil
		}
		if err := writer.Write(batch); err != nil {
			return fmt.Errorf("写入表 %s: %w", name, err)
		}
		written += uint64(len(batch))
		batch = batch[:0]
		if time.Since(lastRep) >= progressEvery {
			lastRep = time.Now()
			o.progress(name, written, time.Since(begin))
		}
		return nil
	}

	scanErr := src.Scan(name, sc, func(r Row) error {
		if o.Limit > 0 && written+uint64(len(batch)) >= o.Limit {
			return errStop
		}
		batch = append(batch, r)
		if o.EachRow != nil {
			o.EachRow(name, written+uint64(len(batch)), r)
		}
		if len(batch) >= o.Batch {
			return flush()
		}
		return nil
	})
	if scanErr != nil && !errors.Is(scanErr, errStop) {
		abortWriter(writer)
		tr.Rows = written
		return tr, fmt.Errorf("扫描表 %s: %w", name, scanErr)
	}
	if err := flush(); err != nil {
		abortWriter(writer)
		tr.Rows = written
		return tr, err
	}
	if err := writer.Close(); err != nil {
		tr.Rows = written
		return tr, fmt.Errorf("收尾表 %s: %w", name, err)
	}
	// 每张表落一次盘: 中途出错前面迁完的不会白费。
	if c, ok := dst.(Committer); ok {
		if err := c.Commit(); err != nil {
			tr.Rows = written
			return tr, fmt.Errorf("提交表 %s: %w", name, err)
		}
	}

	tr.Rows = written
	tr.Elapsed = time.Since(begin)
	if o.Limit > 0 && written >= o.Limit {
		tr.Note = appendNote(tr.Note, fmt.Sprintf("截断到 %d 行", o.Limit))
	}
	o.vlogf("[migrate] %s → %s 表 %s: %d 行 %d 列 %d 索引 (%s)",
		src.Kind(), dst.Kind(), name, tr.Rows, tr.Columns, tr.Indexes, tr.Elapsed.Round(time.Millisecond))
	return tr, nil
}

// appendNote 往报告备注里续一条 (分号分隔), 已有内容时追加而不是覆盖。
func appendNote(note, add string) string {
	if note == "" {
		return add
	}
	return note + "; " + add
}

// prepareForVRTX 把一张(可能来自 SQL 的) schema 收拾成 VRTX 能建表的样子。
//
// VRTX 的行数据寄存在"行存储索引"(主键/时序)的条目里, 而这个索引的键必须能承载
// 行号 —— 落到 SQL 上就是"主键只能是 8 字节整数列"。所以:
//   - 行存储索引的键列已由 __vrtx_meta 指明 (时序表) → 不用动;
//   - 主键已经是 uint64 列 → 什么都不用做;
//   - 主键是字符串/其它宽度 → 补一个 _id 列承载行号, 原主键列退化成普通列
//     (顺手给它建个 B+ 树索引, 找回"按原主键查"的能力);
//   - 压根没有主键 → 只有 AutoPK 打开才补 _id, 否则报错 (VRTX 不接受没有行主键的表)。
//
// 后两条都是【改结构】, 所以会往 sc.DegradedColumns 记一笔由 runTable 报给用户;
// 复合主键 (源表有主键但承载不了) 则记在 sc.PKNote 上, 报错时也用它说准话。
func prepareForVRTX(sc *TableSchema, o *Options) error {
	if len(sc.RowStoreColumns) > 0 {
		// 时序表: 行存储索引建在时间列上 (RowStore), 行号由引擎分配, 不占列。
		return nil
	}
	if sc.PKColumn != "" {
		// 已有的主键列若是 8 字节整数, 钉到 uint64 档即可直接承载行号 (SQL 的整数会先被
		// 收编到有符号的 Int64 档, 见 rowStorePKType); 窄列/非整数列才需要补 _id。
		if pinRowStorePK(sc, sc.PKColumn) {
			return nil
		}
	} else if !o.AutoPK {
		if sc.PKNote != "" {
			// 源表【有】主键, 只是多列 —— 报"没有主键"是假消息, 用户会去源库找不存在的问题。
			return fmt.Errorf("表 %s: %s —— 请改用单列 8 字节整数主键, "+
				"或开 AutoPK 让它自动补 %s 列", sc.Name, sc.PKNote, "_id")
		}
		return fmt.Errorf("表 %s 没有主键, 目标 VRTX 的表必须有行主键列 —— "+
			"请给源表加 8 字节整数主键, 或开 AutoPK 让它自动补 %s 列", sc.Name, "_id")
	}

	origPK := sc.PKColumn
	idName := freeColumnName(sc, "_id")
	sc.Columns = append(sc.Columns, Column{Name: idName, Type: types.FieldType_Uint64})
	sc.MarkSynthetic(idName)
	sc.PKColumn = idName
	sc.PKIndexType = page.IndexTypePrimary
	sc.PKIndexName = "" // 让 VRTX 侧按默认 pk_<表名> 命名

	if origPK == "" {
		return nil
	}
	// 原主键列不再是行主键, 但"按它查"是源库的既有用法, 用二级索引续上。
	ensureColumnIndex(sc, origPK)
	// 记下源库的主键列名: 行号已经换成引擎分配的 _id, 只有这个列名能把写失败的行报回源库
	// 认得的身份 (见 vrtxWriter.rowTag 与 TableSchema.SourcePKColumn)。
	sc.SourcePKColumn = origPK
	// 这一步是【约束降级】: 源库的主键唯一性由主键约束强制, 而补出来的二级索引不是唯一
	// 索引。静默丢掉一个约束比报错更糟 —— 用户会以为主键还在, 继续依赖它。
	sc.DegradedColumns = append(sc.DegradedColumns, fmt.Sprintf(
		"主键 %s 承载不了行号 (不是 8 字节整数列): 已改用引擎分配的 %s 列, "+
			"原主键列降为普通列 + 普通索引 (源库的主键唯一性不再强制)", origPK, idName))
	return nil
}

// ensureColumnIndex 给某一列补一个 B+ 树二级索引 (已有同列索引则跳过)。
//
// 主键列退化成普通列时用它兜底: 少了主键的"按列查"能力, 至少让查询还能走索引。
//
// 动态列 / blob 列不补: 它们的行内只有引用 (堆页号 / 链引用), 按值查不到, 建了也会被
// core.CreateIndex 拒 (ErrIndexDynamicField / ErrIndexBlobField) —— 与其让整张表的迁移
// 因此失败, 不如干脆不补这一个索引。
func ensureColumnIndex(s *TableSchema, col string) {
	if i, ok := s.ColumnIdx(col); ok {
		if c := s.Columns[i]; c.IsDynamic() || c.Type == types.FieldType_Blob {
			return
		}
	}
	for _, idx := range s.Indexes {
		if idx.SameColumns([]string{col}) {
			return
		}
	}
	s.Indexes = append(s.Indexes, IndexSchema{
		Name:    indexNameFor(s.Name, col),
		Type:    page.IndexTypeBTree,
		Columns: []string{col},
	})
}

// indexNameFor 给"自动补出来的索引"起个不会超长的名字。
// VRTX 索引名有长度上限, 表名本身就可能占掉大半, 因此超长时退化成短名。
func indexNameFor(table, col string) string {
	name := "idx_" + table + "_" + col
	if len(name) <= core.IndexNameSize {
		return name
	}
	// 超长时退化成短名: 保住"能建出来"比名字好看重要。
	sum := uint64(1469598103934665603) // FNV-1a 偏移
	for i := 0; i < len(name); i++ {
		sum ^= uint64(name[i])
		sum *= 1099511628211
	}
	short := fmt.Sprintf("idx_%06x", sum&0xFFFFFF)
	if len(short) <= core.IndexNameSize {
		return short
	}
	return short[:core.IndexNameSize]
}

// ============================ 便捷入口 ============================

// ParseDSN 把命令行参数翻译成一个 Endpoint (既能当源也能当目标)。
//
// 前缀决定库的种类, 后面整段就是它的标识:
//
//	vrtx:/data/app.db
//	sqlite:/data/app.db            也可以是 file:... URI
//	mysql:root:pw@tcp(127.0.0.1:3306)/db?charset=utf8mb4
//
// 返回的对象已经打开, 调用方负责 Close。
func ParseDSN(dsn string, opts *Options) (*Endpoint, error) {
	kind, rest, ok := strings.Cut(dsn, ":")
	if !ok {
		return nil, fmt.Errorf("DSN %q 缺少前缀, 应形如 vrtx:/path 或 sqlite:/path 或 mysql:user:pw@tcp(host)/db", dsn)
	}
	switch Kind(strings.ToLower(kind)) {
	case KindVRTX:
		v, err := OpenVRTX(rest, opts)
		if err != nil {
			return nil, err
		}
		return &Endpoint{Kind: KindVRTX, Src: v, Dst: v}, nil
	case KindSQLite:
		s, err := OpenSQLite(rest, opts)
		if err != nil {
			return nil, err
		}
		return &Endpoint{Kind: KindSQLite, Src: s, Dst: s}, nil
	case KindMySQL:
		s, err := OpenMySQL(rest, opts)
		if err != nil {
			return nil, err
		}
		return &Endpoint{Kind: KindMySQL, Src: s, Dst: s}, nil
	}
	return nil, fmt.Errorf("DSN %q 的前缀 %q 不是 vrtx/sqlite/mysql", dsn, kind)
}

// Endpoint 解析出来的一个库: 同时能当源和目标用。
// 两个字段指向同一个底层对象, 只是按接口名区分, 让调用点读起来清楚。
type Endpoint struct {
	Kind Kind
	Src  Source
	Dst  Sink
}

// Identity 该库的唯一标识 (路径 / DSN)。
func (e *Endpoint) Identity() string {
	if e.Src != nil {
		return e.Src.Identity()
	}
	if e.Dst != nil {
		return e.Dst.Identity()
	}
	return ""
}

// Close 关掉底层连接。
func (e *Endpoint) Close() error {
	if e.Dst != nil {
		return e.Dst.Close()
	}
	if e.Src != nil {
		return e.Src.Close()
	}
	return nil
}