TsDB2/migrate/vrtx.go

package migrate

import (
	"TsDB2/DB/core"
	"TsDB2/DB/core/page"
	"TsDB2/DB/core/types"
	"TsDB2/DB/core/util"
	"errors"
	"fmt"
	"sort"
	"strings"
)

// VRTX 一侧。同一个对象既能当 Source 也能当 Sink (方向由调用方决定), 于是任意
// 方向 (VRTX→SQL / SQL→VRTX / VRTX→VRTX) 都只需要这一套读写实现。
//
// 这里刻意走 core 层而不是 ORM (package DB): 表结构在迁移时来自源库的元数据
// (列名 / 类型 / 行存储索引列), 运行时才知道, 根本没有一个 Go 结构体可以内嵌表基类 ——
// 而 ORM 的入口 UseTable[T] 恰恰要求 T 内嵌 DB.BaseModel / DB.TimeSeriesModel
// (见 DB 包的 requireBase)。core 层只认 []types.FieldInfo, 没有"先有结构体"这个前提,
// 所以"按库里的元数据建表"只能落在 core 层。ORM 是给"写代码时就知道表结构"的场景用的。
type VRTX struct {
	db   *core.DB
	path string
	opts *Options
}

// OpenVRTX 打开 (不存在则新建) 一个 VRTX 库文件。
func OpenVRTX(path string, opts *Options) (*VRTX, error) {
	db, err := core.Open(path)
	if err != nil {
		return nil, fmt.Errorf("打开 VRTX 库 %s: %w", path, err)
	}
	return &VRTX{db: db, path: path, opts: opts.normalized()}, nil
}

func (v *VRTX) Kind() Kind       { return KindVRTX }
func (v *VRTX) Identity() string { return "vrtx:" + v.path }

// DB 暴露底层句柄 (调用方可能需要调缓冲池 / 手动提交)。
func (v *VRTX) DB() *core.DB { return v.db }

// Commit 把当前脏页落盘。编排器每迁完一张表调一次: VRTX 的插入是攒在缓冲池里的,
// 千万行的库若要等最后 Close 才提交, 中途崩一次就是全部白干, 内存也会一直涨。
func (v *VRTX) Commit() error { return v.db.Commit() }

// Close 落盘并关闭。
func (v *VRTX) Close() error {
	// 迁移是批量写, 收尾必须显式提交一次: 否则最后一批脏页只剩防抖定时器兜底,
	// 进程立刻退出就会丢。
	if err := v.db.Commit(); err != nil {
		_ = v.db.Close()
		return err
	}
	return v.db.Close()
}

// Tables 列出全部表 (core 已跳过软删除的表)。
func (v *VRTX) Tables() ([]string, error) {
	metas := v.db.GetAllTables()
	names := make([]string, 0, len(metas))
	for _, m := range metas {
		names = append(names, m.Name)
	}
	sort.Strings(names) // 稳定顺序: 迁移过程可复现、可对照
	return names, nil
}

// Schema 读一张表的结构。
//
// 列取 master registry 里未被软删除的那些 —— 注意用 master 顺序而不是
// GetActiveFieldNames(): 后者是"当前 schema 版本"的列序, 而 master registry 的
// 下标才是索引项里 FieldIDs 引用的那个下标, 换算列名必须用它。
func (v *VRTX) Schema(table string) (*TableSchema, error) {
	tbl, ok := v.db.FindTable(table)
	if !ok {
		return nil, fmt.Errorf("VRTX 库 %s 里没有表 %s", v.path, table)
	}
	fields := tbl.GetFields() // master registry: 含已软删除的列 (删除标记见 types.FlagDeleted)

	s := &TableSchema{Name: table}
	for _, f := range fields {
		if f.IsDropped() {
			continue
		}
		s.Columns = append(s.Columns, Column{
			Name:  util.BytesToString(f.FieldName),
			Type:  f.FieldType,
			Param: f.Flag & types.FlagScaleMask, // Decimal 的列级 scale, 必须跟着列走
			// NULL / 唯一声明: 前者决定导出时写不写 NOT NULL, 后者是列级声明
			// (约束本身由索引执行, 见下面 GetIndexes 的 m.Unique())。
			Nullable: f.IsNullable(),
			Unique:   f.IsUnique(),
		})
	}

	im := tbl.IndexMeta()
	pk, hasPK := im.GetRowStoreIndex()
	if !hasPK {
		return nil, fmt.Errorf("VRTX 表 %s 没有行存储索引 (主键/时序), 行数据无处安放", table)
	}
	s.PKIndexName = pk.Name()
	s.PKIndexType = pk.IndexType

	pkCols := make([]string, 0, pk.FieldCnt)
	for _, fid := range pk.Fields() {
		if int(fid) >= len(fields) {
			continue
		}
		if util.IsHighBitSet(byte(fields[fid].FieldType)) {
			continue
		}
		pkCols = append(pkCols, util.BytesToString(fields[fid].FieldName))
	}
	switch len(pkCols) {
	case 0:
		// 行存储索引的分量列全被删了: 保留 PKColumn 为空, 由下面补列
	case 1:
		s.PKColumn = pkCols[0]
	default:
		return nil, fmt.Errorf("VRTX 表 %s 的行存储索引 %s 有 %d 个分量, 不是合法的行存储索引",
			table, pk.Name(), len(pkCols))
	}

	for _, m := range im.GetIndexes() {
		if m.Name() == s.PKIndexName {
			continue // 主键索引由 PKColumn / PKIndexType 表达
		}
		cols := make([]string, 0, m.FieldCnt)
		broken := false
		for _, fid := range m.Fields() {
			if int(fid) >= len(fields) || util.IsHighBitSet(byte(fields[fid].FieldType)) {
				broken = true // 分量列已被删除: 索引名存实亡, 导过去只会是个坏索引
				break
			}
			cols = append(cols, util.BytesToString(fields[fid].FieldName))
		}
		if broken || len(cols) == 0 {
			v.opts.logf("[vrtx] 表 %s 的索引 %s 有分量列已被删除, 跳过", table, m.Name())
			continue
		}
		s.Indexes = append(s.Indexes, IndexSchema{
			Name:    m.Name(),
			Type:    m.IndexType,
			Columns: cols,
			Unique:  m.Unique(), // IndexMeta.Flag 的唯一位: 约束的执行者
		})
	}

	// 时序索引: 行存储的键是"时间列", 与行号无关 —— 既不能当 SQL 主键 (主键得是整数),
	// 也不能把行号往时间列里写。这类表导出到 SQL 时不设主键, 时间列照常是普通列;
	// 行存储索引的类型和键列记在 __vrtx_meta 里, 回来时原样重建。
	//
	// 顺带说明为什么不给时间列补 _id 或普通索引: 时序索引自己维护的 pkID 并不可靠
	// (首个分段落盘时会把首列数据当成 pkID 存, 见 datapage.CreateDataPage), 所以这条
	// 路径上干脆不读它, 让目标端按自增号分配; 时间列是行存储键, 建普通索引也是重复。
	if s.PKIndexType == page.IndexTypeTime && s.PKColumn != "" &&
		pinRowStorePK(s, s.PKColumn) {
		s.RowStoreColumns = append([]string(nil), pkCols...)
		s.PKColumn = ""
		return s, nil
	}

	// 其余情况行主键必须能承载行号 (SQL 的 PRIMARY KEY 只能是整数, VRTX 侧这个列
	// 也是 pkID 的镜像)。行存储索引挂在定长串这类非 8 字节列上时, 该列的值与行号
	// 不是一回事 —— 补一个 _id 列专门承载行号, 原列退化成普通列。
	//
	// pinRowStorePK 顺带把"有符号 8 字节整数"钉回 8byte 档 (见 rowStorePKType)。
	if !pinRowStorePK(s, s.PKColumn) {
		name := freeColumnName(s, "_id")
		s.Columns = append(s.Columns, Column{Name: name, Type: types.FieldType_8byte})
		s.MarkSynthetic(name)
		if s.PKColumn != "" {
			v.opts.logf("[vrtx] 表 %s 的行存储索引挂在 %s (%s) 上, 不是 8 字节整数列, "+
				"补列 %s 承载行号", table, s.PKColumn, TypeName(typeOfColumn(s, s.PKColumn)), name)
			// 原列变成普通列, 不参与主键; 但"按它查"是原来的用法, 补个索引续上
			ensureColumnIndex(s, s.PKColumn)
		}
		s.PKIndexType = page.IndexTypePrimary
		s.PKColumn = name
	}
	return s, nil
}

// Scan 全表扫描。回调拿到的 Row.Values 与 s.Columns 同序。
func (v *VRTX) Scan(table string, s *TableSchema, fn func(Row) error) error {
	tbl, ok := v.db.FindTable(table)
	if !ok {
		return fmt.Errorf("VRTX 库 %s 里没有表 %s", v.path, table)
	}
	pos := make(map[string]int, len(s.Columns))
	for i, c := range s.Columns {
		pos[c.Name] = i
	}
	pkIdx := columnIndex(s, s.PKColumn)
	diverged := false
	var fnErr error

	scanErr := tbl.Query().ScanWithPKID(func(pkID uint64, vals []types.FieldValue) bool {
		row := Row{PK: pkID, HasPK: true, Values: make([]any, len(s.Columns))}
		for _, fv := range vals {
			i, ok := pos[util.BytesToString(fv.FieldName)]
			if !ok {
				continue // 已删除的列 / 不在导出范围内的列
			}
			row.Values[i] = fv.Value
		}
		// 主键列以行号为准: 它本来就是 pkID 的镜像。真有对不上的 (绕过 ORM 直接
		// 用 core 写入、主键列给了别的值), 说一声, 免得以为是导错了。
		if pkIdx >= 0 {
			if old := row.Values[pkIdx]; old != nil {
				if n, err := asUint64(old); err == nil && n != pkID && !diverged {
					diverged = true
					v.opts.logf("[vrtx] 表 %s 有行的主键列 %s 与行号不一致 (如 %v ≠ %d), "+
						"导出时以行号为准", table, s.PKColumn, old, pkID)
				}
			}
			row.Values[pkIdx] = pkID
		}
		if err := fn(row); err != nil {
			fnErr = err
			return false
		}
		return true
	})
	if fnErr != nil {
		return fnErr
	}
	if scanErr != nil {
		return fmt.Errorf("扫描 VRTX 表 %s: %w", table, scanErr)
	}
	return nil
}

// ============================ Sink ============================

// Exists 目标表是否存在。
func (v *VRTX) Exists(table string) (bool, error) {
	_, ok := v.db.FindTable(table)
	return ok, nil
}

// CreateTable 建表 + 建索引 (主键索引必须建, 否则行数据无处安放)。
func (v *VRTX) CreateTable(s *TableSchema, recreate bool) error {
	exists, err := v.Exists(s.Name)
	if err != nil {
		return err
	}
	if exists && recreate {
		if err := v.db.DropTable(s.Name); err != nil {
			return fmt.Errorf("DROP VRTX 表 %s: %w", s.Name, err)
		}
		exists = false
	}
	if exists {
		return nil // 复用已有表: 列与索引保持原样, 数据追加
	}
	if len(s.RowStore()) == 0 {
		return fmt.Errorf("VRTX 表 %s 没有行存储索引的列 (主键列为空), 无法建表", s.Name)
	}
	if len(s.Name) > core.TableNameSize {
		return fmt.Errorf("表名 %s 有 %d 字节, 超过 VRTX 上限 %d", s.Name, len(s.Name), core.TableNameSize)
	}

	// 单列唯一索引 ⇒ 这一列同时是"唯一列", 把列上的声明位补上。
	//
	// 声明与约束是两件事 (列位给读者看, 索引负责拦), 但只有索引这一半能从 SQL 读
	// 出来 —— SQLite 的列级 UNIQUE 在 PRAGMA table_info 里毫无痕迹, 只生成一条
	// autoindex。补位不会凭空多出约束 (约束一直是索引给的), 少了它却会让从外部库
	// 迁进来的表"看不出这列本该唯一"。复合唯一索引不补: (A,B) 唯一不等于 B 唯一。
	for _, idx := range s.Indexes {
		if !idx.Unique || len(idx.Columns) != 1 {
			continue
		}
		if ci := columnIndex(s, idx.Columns[0]); ci >= 0 {
			s.Columns[ci].Unique = true
		}
	}

	tbl, err := v.db.CreateTable(s.Name, vrtxFields(s))
	if err != nil {
		return fmt.Errorf("建 VRTX 表 %s: %w", s.Name, err)
	}

	rsCols := s.RowStore()
	if len(rsCols) == 0 {
		return fmt.Errorf("表 %s 没有行存储索引的列 (主键列为空)", s.Name)
	}
	pkType := s.PKIndexType
	if pkType != page.IndexTypeTime {
		pkType = page.IndexTypePrimary
	}
	ids := make([]byte, 0, len(rsCols))
	for _, col := range rsCols {
		ci := columnIndex(s, col)
		if ci < 0 {
			return fmt.Errorf("表 %s 的行存储索引列 %s 不在列清单里", s.Name, col)
		}
		if pkType == page.IndexTypeTime && s.Columns[ci].Type != types.FieldType_8byte {
			// 时序索引的键就是 8 字节时间戳, 宽度不对会在写入时直接崩。
			return fmt.Errorf("表 %s 的行存储索引列 %s 是 %s, 时序索引要求 8 字节整数列",
				s.Name, col, TypeName(s.Columns[ci].Type))
		}
		ids = append(ids, byte(ci))
	}
	pkName := s.PKIndexName
	if pkName == "" {
		pkName = "pk_" + s.Name
	}
	if err := checkIndexName(pkName); err != nil {
		return err
	}
	if _, err := tbl.CreateIndex(pkName, pkType, ids); err != nil {
		return fmt.Errorf("建主键索引 %s: %w", pkName, err)
	}

	for _, idx := range s.Indexes {
		if idx.Name == pkName || idx.IsRowStore() {
			continue
		}
		// 索引名超长 / 分量过多: 与下面的"动态列当键"同一处理 —— 跳过并记进报告。
		// 这两种都是"源库合法、VRTX 装不下" (SQLite 的索引名与列数几乎没有限制), 但都
		// 只影响这条索引, 不该让整张表的数据陪葬。以前这里直接 return err, 于是一个
		// 名字太长的索引能把整张表拦下来。
		if bad := auditIndexForVRTX(idx); bad != "" {
			s.SkippedIndexes = append(s.SkippedIndexes, idx.Name+": "+bad)
			v.opts.logf("[vrtx] 表 %s 的索引 %s 建不了, 跳过 (%s)", s.Name, idx.Name, bad)
			continue
		}
		ids := make([]byte, 0, len(idx.Columns))
		bad := ""
		for _, col := range idx.Columns {
			ci := columnIndex(s, col)
			if ci < 0 {
				bad = "列 " + col + " 不在表中"
				break
			}
			if s.Columns[ci].IsDynamic() {
				// 源库里它是条合法索引 (MySQL 的长 varchar / text 都能建), 但 VRTX
				// 行内只存动态列的堆引用, 按值查不到, 引擎会直接拒 (ErrIndexDynamicField)。
				// 跳过这个索引, 而不是让整张表失败 —— 索引丢了还能补, 数据没迁才是真丢。
				bad = fmt.Sprintf("键列 %s 是 %s 动态列, VRTX 不能拿它当索引键",
					col, TypeName(s.Columns[ci].Type))
				break
			}
			ids = append(ids, byte(ci))
		}
		if bad != "" {
			s.SkippedIndexes = append(s.SkippedIndexes, idx.Name+": "+bad)
			v.opts.logf("[vrtx] 表 %s 的索引 %s 建不了, 跳过 (%s)", s.Name, idx.Name, bad)
			continue
		}
		idxType := idx.Type
		if idxType == 0 || idxType == page.IndexTypePrimary || idxType == page.IndexTypeTime {
			idxType = page.IndexTypeBTree
		}
		if _, err := tbl.CreateIndex(idx.Name, idxType, ids, idx.Unique); err != nil {
			return fmt.Errorf("建索引 %s: %w", idx.Name, err)
		}
	}
	return nil
}

// Open 准备写入。
func (v *VRTX) Open(table string, s *TableSchema) (RowWriter, error) {
	tbl, ok := v.db.FindTable(table)
	if !ok {
		return nil, fmt.Errorf("VRTX 库 %s 里没有表 %s (要先建表)", v.path, table)
	}
	return &vrtxWriter{tbl: tbl, schema: s, opts: v.opts, pkIdx: columnIndex(s, s.PKColumn)}, nil
}

type vrtxWriter struct {
	tbl    *core.Table
	schema *TableSchema
	opts   *Options
	pkIdx  int
}

// Write 逐行写入。VRTX 没有"批量插入"接口 (每行都要维护 B+ 树与索引项),
// 所以这里就是 for 循环 —— 批的意义在于 SQL 侧的往返次数。
func (w *vrtxWriter) Write(rows []Row) error {
	vals := make([]types.FieldValue, len(w.schema.Columns))
	for i, c := range w.schema.Columns {
		vals[i].FieldName = []byte(c.Name)
	}
	for ri := range rows {
		r := &rows[ri]
		pkID := r.PK
		if !r.HasPK || pkID == 0 {
			pkID = w.tbl.AllocPKID()
		}
		for i, c := range w.schema.Columns {
			var raw any
			if i < len(r.Values) {
				raw = r.Values[i]
			}
			if i == w.pkIdx {
				raw = pkID // 主键列是行号的镜像
			}
			cv, err := coerceForVRTX(c.Type, c.Param, raw, c.Name, w.opts.StrictValue, c.Nullable)
			if err != nil {
				return err
			}
			vals[i].Value = cv
		}
		if _, err := w.tbl.InsertRowWithID(pkID, vals); err != nil {
			return fmt.Errorf("写入 VRTX 表 %s: %w", w.schema.Name, err)
		}
	}
	return nil
}

// Close 提交这一批 (VRTX 的提交由 DB 层统一做, 这里只留扩展点)。
func (w *vrtxWriter) Close() error { return nil }

// ============================ 工具 ============================

func checkIndexName(name string) error {
	if name == "" {
		return errors.New("索引名为空")
	}
	if len(name) > core.IndexNameSize {
		return fmt.Errorf("索引名 %s 有 %d 字节, 超过 VRTX 上限 %d (换短一点的名字)",
			name, len(name), core.IndexNameSize)
	}
	return nil
}

// columnFlag 列的 VRTX 标志位: 低 5 位 scale + NULL / 唯一声明。
//
// 三者共用一个字节 (布局见 types 的 Flag* 常量), 少写任何一位都是静默改结构:
// 丢了 FlagNullable, 往这列写 NULL 会被当成"给个零值"塞 0; 丢了 FlagUnique,
// 反向推 schema 的工具 (以及 ORM 的 tag) 就看不出这列本该唯一。
func columnFlag(c Column) byte {
	f := c.Param & types.FlagScaleMask
	if c.Nullable {
		f |= types.FlagNullable
	}
	if c.Unique {
		f |= types.FlagUnique
	}
	return f
}

// vrtxFields 把列清单翻成 core 建表要的 []types.FieldInfo。
//
// 抽出来是因为它有两个用途, 两处必须逐字节一致 —— 建表用它, 行宽闸门 (更宽的行装不进
// 一页两行) 也用它算行宽 (见 fitRowWidthForVRTX)。各写一份的话, 一旦某天只改了建表那份
// (比如新增一个列标志位), 行宽闸门算的就是另一张表: 闸门放过的表建不出来。
func vrtxFields(s *TableSchema) []types.FieldInfo {
	fields := make([]types.FieldInfo, 0, len(s.Columns))
	for _, c := range s.Columns {
		fields = append(fields, types.FieldInfo{
			FieldType: c.Type,
			FieldName: []byte(c.Name),
			Flag:      columnFlag(c), // decimal scale + NULL / 唯一声明
		})
	}
	return fields
}

func columnIndex(s *TableSchema, name string) int {
	if i, ok := s.ColumnIdx(name); ok {
		return i
	}
	return -1
}

func typeOfColumn(s *TableSchema, name string) types.FieldType {
	if i, ok := s.ColumnIdx(name); ok {
		return s.Columns[i].Type
	}
	return 0
}

// rowStorePKType 判断某列能否直接承载行号 (pkID 的镜像), 并给出它在 VRTX 侧必须使用的类型。
//
// 行存储索引的键就是一个 8 字节整数列: 它的值即引擎行主键号, 主键 B+ 树按它的字节序排列;
// 时序表里这一列同时承载时间戳, 时序索引按 int64 补码直读这 8 个字节。因此它在 VRTX 侧
// 必须落在【无符号 8byte 档】上 —— 盘上就是补码本身。
//
// 而 SQL 的整数默认被 inferSQLType 收编到有符号档 (Int64, 盘上是保序编码): 宽度同为 8 字节,
// 但字节布局不同, 直接拿去当行主键会让主键树错序、当时间键会让时间轴错位。这里把它钉回
// 8byte; 行号本身非负, 值不会因此改变。1/2/4 字节的整数列装不下行号, 返回 false,
// 由调用方补 _id 列。
func rowStorePKType(t types.FieldType) (types.FieldType, bool) {
	switch t {
	case types.FieldType_8byte:
		return types.FieldType_8byte, true
	case types.FieldType_Int64:
		return types.FieldType_8byte, true
	}
	return 0, false
}

// pinRowStorePK 就地把指定列钉到行存储主键要求的档位; 列不存在或宽度不够时返回 false。
func pinRowStorePK(s *TableSchema, col string) bool {
	i, ok := s.ColumnIdx(col)
	if !ok {
		return false
	}
	ft, ok := rowStorePKType(s.Columns[i].Type)
	if !ok {
		return false
	}
	s.Columns[i].Type = ft
	return true
}

// freeColumnName 找一个没被占用的列名 (优先用 want)。
func freeColumnName(s *TableSchema, want string) string {
	if !s.HasColumn(want) {
		return want
	}
	for i := 2; ; i++ {
		cand := fmt.Sprintf("%s_%d", want, i)
		if !s.HasColumn(cand) {
			return cand
		}
	}
}

// vrtxTableName 把外部表名收拾成 VRTX 能接受的样子 (名字里有空格/奇怪字符时)。
func vrtxTableName(name string) string {
	return strings.TrimSpace(name)
}