package migrate

import (
	"git.ycl.pub/yl32940/TsDB2/DB/core/page"
	"git.ycl.pub/yl32940/TsDB2/DB/core/types"
	"database/sql"
	"encoding/json"
	"errors"
	"fmt"
	"sort"
	"strings"
)

// MetaTable 保真表的名字。它在目标库里只是一张普通表, 别的工具读到也不影响;
// 删掉它, 迁移依然能跑 (类型/索引类型按 SQL 声明就近推断), 只是定长档位与
// Hash/时序这些 SQL 表达不了的细节会丢。
const MetaTable = "__vrtx_meta"

// dialect SQL 方言。MySQL 与 SQLite 只有这些差异, 读写流程完全共用。
// nullClause 建表时的可空声明。
//
// 只有"不可空"才写 NOT NULL —— 可空列不能靠"什么都不写"来表达吗? 能, 但显式
// 写一个 NULL 更好: 生成出来的 DDL 一眼能看出这列是【有意可空】, 而不是从
// VRTX 导过来时被漏掉了可空声明 (两者的 DDL 长得一样, 只有写了 NULL 才区分得开)。
func nullClause(c Column) string {
	if c.Nullable {
		return " NULL"
	}
	return " NOT NULL"
}

// createIndexKind 索引 DDL 的动词前缀。唯一索引在 SQL 侧就是 UNIQUE INDEX,
// 没有别的表达方式 —— 少了它, VRTX 的唯一约束导出到 SQL 就只剩普通索引。
func createIndexKind(idx IndexSchema) string {
	if idx.Unique {
		return "CREATE UNIQUE INDEX"
	}
	return "CREATE INDEX"
}

type dialect interface {
	kind() Kind
	quote(name string) string

	// createTable 生成含主键的 CREATE TABLE 语句 (二级索引另建)。
	createTable(s *TableSchema) string
	dropTable(table string) string
	createIndex(table string, idx IndexSchema) string
	// columnDDL 单列的列定义 (名字 + 类型)。
	columnDDL(c Column) string
	// addColumn 给已有表补列的语句 (复用已有表时用)。
	addColumn(table string, c Column) string

	// listTables 列出用户表 (不含内部表)。
	listTables(db *sql.DB) ([]string, error)
	// columns 读列 (按表内顺序)、主键列名, 以及"源库的结构 VRTX 承载不了"的说明。
	//
	// 第三个返回值是读端才看得见的东西 (只有这里拿得到 SQL 的声明原文与主键构成),
	// 由 SQL.Schema 落到 TableSchema 的 CappedColumns / DegradedColumns / PKNote 上,
	// 最终进迁移报告 —— 见 sourceNotes。
	columns(db *sql.DB, table string) ([]Column, string, sourceNotes, error)
	// indexes 读二级索引 (不含主键索引)。
	indexes(db *sql.DB, table string) ([]IndexSchema, error)

	// metaDDL 保真表建表语句。
	metaDDL() string
	// metaUpsert 写一行保真数据的语句 (两个 ? 占位: 表名, payload)。
	metaUpsert() string
	// metaSelect 读一张表的保真数据。
	metaSelect() string

	// maxPlaceholders 单条语句允许的最大占位符数量 (决定多行 INSERT 一次插几行)。
	maxPlaceholders() int

	// textLenExpr "这一列占多少字节"的表达式 (喂给 MAX(...) 做 TEXT 采样定宽),
	// col 是【未引用】的列名, 实现内部按本方言引用。两方的 length 语义不同:
	// SQLite 的 length() 对文本数的是字符, MySQL 的 LENGTH() 本身就是字节数。
	textLenExpr(col string) string
}

// SQL 一侧 (MySQL / SQLite 共用)。
type SQL struct {
	db       *sql.DB
	kind     Kind
	dsn      string
	dial     dialect
	opts     *Options
	tables   []string
	hasMeta  map[string]*metaDTO
	metaRead map[string]bool
}

func newSQL(db *sql.DB, kind Kind, dsn string, d dialect, opts *Options) *SQL {
	return &SQL{
		db:       db,
		kind:     kind,
		dsn:      dsn,
		dial:     d,
		opts:     opts.normalized(),
		metaRead: map[string]bool{},
	}
}

func (s *SQL) Kind() Kind       { return s.kind }
func (s *SQL) Identity() string { return string(s.kind) + ":" + s.dsn }

// DB 暴露底层句柄。
func (s *SQL) DB() *sql.DB { return s.db }

func (s *SQL) Close() error { return s.db.Close() }

// Tables 列出可迁移的表 (不含保真表; 除非调用方明确点名要它)。
func (s *SQL) Tables() ([]string, error) {
	if s.tables == nil {
		names, err := s.dial.listTables(s.db)
		if err != nil {
			return nil, err
		}
		s.tables = names
	}
	out := make([]string, 0, len(s.tables))
	for _, n := range s.tables {
		if n == MetaTable && !s.opts.wants(MetaTable) {
			continue
		}
		out = append(out, n)
	}
	sort.Strings(out)
	return out, nil
}

// Schema 读一张表的结构: SQL 声明 → VRTX 类型, 再用保真表 (若有) 覆盖回来。
func (s *SQL) Schema(table string) (*TableSchema, error) {
	cols, pk, notes, err := s.dial.columns(s.db, table)
	if err != nil {
		return nil, fmt.Errorf("读 %s 表 %s 的结构: %w", s.kind, table, err)
	}
	if len(cols) == 0 {
		return nil, fmt.Errorf("%s 表 %s 不存在或没有列", s.kind, table)
	}
	schema := &TableSchema{
		Name:            table,
		Columns:         cols,
		PKColumn:        pk,
		PKIndexType:     page.IndexTypePrimary,
		CappedColumns:   notes.Capped,
		DegradedColumns: notes.Degraded,
		PKNote:          notes.PK,
	}
	if idx, err := s.dial.indexes(s.db, table); err == nil {
		schema.Indexes = idx
	} else {
		s.opts.logf("[%s] 读表 %s 的索引列表失败 (忽略): %v", s.kind, table, err)
	}

	// 保真表优先: 列宽档位、索引类型、索引名都按当初导出时记的还原。
	if s.opts.EmbedMeta {
		if m, ok := s.readMeta(table); ok {
			applyMeta(schema, m)
		}
	}
	return schema, nil
}

// Scan 全表扫描 (不带 WHERE, 顺序由存储决定)。
func (s *SQL) Scan(table string, sc *TableSchema, fn func(Row) error) error {
	// 补出来的列 (源 SQL 表里并不存在) 不能出现在 SELECT 里。
	// 同时记下每个被选中的列在 schema 里的下标 —— 扫描结果要按它放回原位,
	// 补列的位置留 nil (由写端自己填)。
	sel := make([]int, 0, len(sc.Columns))
	for i, c := range sc.Columns {
		if sc.IsSynthetic(c.Name) {
			continue
		}
		sel = append(sel, i)
	}
	if len(sel) == 0 {
		return fmt.Errorf("%s 表 %s 没有可读的列", s.kind, table)
	}
	names := make([]string, len(sel))
	for k, i := range sel {
		names[k] = s.dial.quote(sc.Columns[i].Name)
	}
	q := fmt.Sprintf("SELECT %s FROM %s", strings.Join(names, ", "), s.dial.quote(table))
	rows, err := s.db.Query(q)
	if err != nil {
		return fmt.Errorf("查询 %s 表 %s: %w", s.kind, table, err)
	}
	defer rows.Close()

	pkIdx := columnIndex(sc, sc.PKColumn)
	dests := make([]any, len(sel))
	for i := range dests {
		var v any
		dests[i] = &v
	}
	for rows.Next() {
		if err := rows.Scan(dests...); err != nil {
			return fmt.Errorf("读 %s 表 %s 的行: %w", s.kind, table, err)
		}
		// 注意: 驱动给的值只在本次迭代内有效 (可能复用底层缓冲), 回调必须当场消费。
		vals := make([]any, len(sc.Columns))
		for k, i := range sel {
			vals[i] = *(dests[k].(*any))
		}
		row := Row{Values: vals}
		if pkIdx >= 0 && vals[pkIdx] != nil {
			if n, err := asUint64(vals[pkIdx]); err == nil {
				row.PK, row.HasPK = n, true
			} else if s.opts.Verbose {
				s.opts.logf("[%s] 表 %s 有行的主键 %v 不是整数, "+
					"该行将按目标库的自增号插入", s.kind, table, vals[pkIdx])
			}
		}
		if err := fn(row); err != nil {
			return err
		}
	}
	if err := rows.Err(); err != nil {
		return fmt.Errorf("遍历 %s 表 %s: %w", s.kind, table, err)
	}
	return nil
}

// TextWidths 对一张表的若干列采样实际最大字节长度 (migrate.TextWidthSampler 的 SQL 实现)。
//
// 同表多列合进一句 `SELECT MAX(...), MAX(...) ... FROM 表` —— 采样按表只做一次;
// 全 NULL (或空表) 的列采到 0。表名/列名按源方言引用, 取自源 schema 元数据。
func (s *SQL) TextWidths(table string, cols []string) (map[string]int64, error) {
	if len(cols) == 0 {
		return nil, nil
	}
	exprs := make([]string, len(cols))
	for i, c := range cols {
		exprs[i] = "MAX(" + s.dial.textLenExpr(c) + ")"
	}
	q := fmt.Sprintf("SELECT %s FROM %s", strings.Join(exprs, ", "), s.dial.quote(table))
	raw := make([]sql.NullInt64, len(cols))
	dest := make([]any, len(cols))
	for i := range raw {
		dest[i] = &raw[i]
	}
	if err := s.db.QueryRow(q).Scan(dest...); err != nil {
		return nil, fmt.Errorf("采样 %s 表 %s 的文本宽度: %w\nSQL: %s", s.kind, table, err, q)
	}
	out := make(map[string]int64, len(cols))
	for i, c := range cols {
		// 全 NULL / 空表: MAX 无值, 视为 0 (与"最短也要 32 档"的规则合起来,
		// 全 NULL 的 TEXT 列落 32 字节档)。
		out[c] = raw[i].Int64
	}
	return out, nil
}

// Exists 目标表是否已存在。
func (s *SQL) Exists(table string) (bool, error) {
	if s.tables == nil {
		if _, err := s.Tables(); err != nil {
			return false, err
		}
	}
	for _, n := range s.tables {
		if n == table {
			return true, nil
		}
	}
	return false, nil
}

// CreateTable 建表 + 建二级索引 + 写保真表。
func (s *SQL) CreateTable(sc *TableSchema, recreate bool) error {
	exists, err := s.Exists(sc.Name)
	if err != nil {
		return err
	}
	if exists && recreate {
		if _, err := s.db.Exec(s.dial.dropTable(sc.Name)); err != nil {
			return fmt.Errorf("DROP %s 表 %s: %w", s.kind, sc.Name, err)
		}
		exists = false
		s.tables = nil
	}
	if !exists {
		if _, err := s.db.Exec(s.dial.createTable(sc)); err != nil {
			return fmt.Errorf("建 %s 表 %s: %w\nSQL: %s", s.kind, sc.Name, err, s.dial.createTable(sc))
		}
		s.tables = nil
		for _, idx := range sc.Indexes {
			if idx.IsRowStore() || idx.Name == "" {
				continue
			}
			if _, err := s.db.Exec(s.dial.createIndex(sc.Name, idx)); err != nil {
				return fmt.Errorf("建索引 %s: %w\nSQL: %s", idx.Name, err, s.dial.createIndex(sc.Name, idx))
			}
		}
	} else if err := s.ensureColumns(sc); err != nil {
		return err
	}

	if s.opts.EmbedMeta {
		if err := s.writeMeta(sc); err != nil {
			return err
		}
	}
	return nil
}

// ensureColumns 复用已有表时补上缺的列 (SQL 的 ADD COLUMN), 免得 INSERT 撞上未知列。
// VRTX 侧的 schema 同步是同一套思路: 结构向"源"看齐, 但不删目标已有的列。
func (s *SQL) ensureColumns(sc *TableSchema) error {
	// 容量说明只在读整表结构时报一次 (这里是补列的辅助路径, 丢掉即可)。
	cols, _, _, err := s.dial.columns(s.db, sc.Name)
	if err != nil {
		return err
	}
	have := make(map[string]bool, len(cols))
	for _, c := range cols {
		have[c.Name] = true
	}
	missing := 0
	for _, c := range sc.Columns {
		if have[c.Name] {
			continue
		}
		// 主键列不能后补 (SQL 里加主键是另一件事, 而且要重建表), 缺了直接报错
		if c.Name == sc.PKColumn {
			return fmt.Errorf("目标表 %s 里没有主键列 %s, 无法追加", sc.Name, c.Name)
		}
		if _, err := s.db.Exec(s.dial.addColumn(sc.Name, c)); err != nil {
			return fmt.Errorf("给表 %s 补列 %s: %w", sc.Name, c.Name, err)
		}
		missing++
	}
	if missing > 0 {
		s.opts.logf("[%s] 表 %s 补了 %d 个缺列", s.kind, sc.Name, missing)
	}
	return nil
}

// Open 准备批量写: 拼多行 INSERT (MySQL/SQLite 的往返次数直接决定导入速度)。
func (s *SQL) Open(table string, sc *TableSchema) (RowWriter, error) {
	n := len(sc.Columns)
	if n == 0 {
		return nil, fmt.Errorf("表 %s 没有列", table)
	}
	rowsPerStmt := s.dial.maxPlaceholders() / n
	if rowsPerStmt < 1 {
		rowsPerStmt = 1
	}
	if rowsPerStmt > s.opts.Batch {
		rowsPerStmt = s.opts.Batch
	}
	commitEvery := uint64(s.opts.Batch) * 8
	if commitEvery < 1000 {
		commitEvery = 1000
	}
	tx, err := s.db.Begin()
	if err != nil {
		return nil, fmt.Errorf("开事务: %w", err)
	}
	return &sqlWriter{
		s:           s,
		table:       table,
		cols:        sc.Columns,
		rowsPerStmt: rowsPerStmt,
		commitEvery: commitEvery,
		tx:          tx,
		stmts:       map[int]*sql.Stmt{},
	}, nil
}

type sqlWriter struct {
	s           *SQL
	table       string
	cols        []Column
	rowsPerStmt int
	commitEvery uint64
	tx          *sql.Tx
	stmts       map[int]*sql.Stmt
	pending     []any
	pendingRows int
	written     uint64
	closed      bool
}

func (w *sqlWriter) insertSQL(rows, cols int) string {
	one := "(" + strings.TrimSuffix(strings.Repeat("?,", cols), ",") + ")"
	vals := strings.TrimSuffix(strings.Repeat(one+",", rows), ",")
	names := make([]string, cols)
	for i, c := range w.cols {
		names[i] = w.s.dial.quote(c.Name)
	}
	return fmt.Sprintf("INSERT INTO %s (%s) VALUES %s",
		w.s.dial.quote(w.table), strings.Join(names, ", "), vals)
}

func (w *sqlWriter) Write(rows []Row) error {
	for ri := range rows {
		r := &rows[ri]
		for i, c := range w.cols {
			var raw any
			if i < len(r.Values) {
				raw = r.Values[i]
			}
			v, err := coerceForSQL(c, raw, w.s.kind == KindMySQL)
			if err != nil {
				return err
			}
			w.pending = append(w.pending, v)
		}
		w.pendingRows++
		if w.pendingRows >= w.rowsPerStmt {
			if err := w.flush(); err != nil {
				return err
			}
		}
	}
	return nil
}

func (w *sqlWriter) flush() error {
	if w.pendingRows == 0 {
		return nil
	}
	stmt, ok := w.stmts[w.pendingRows]
	if !ok {
		s, err := w.tx.Prepare(w.insertSQL(w.pendingRows, len(w.cols)))
		if err != nil {
			return fmt.Errorf("准备 INSERT (表 %s): %w", w.table, err)
		}
		w.stmts[w.pendingRows] = s
		stmt = s
	}
	if _, err := stmt.Exec(w.pending...); err != nil {
		return fmt.Errorf("写入 %s 表 %s: %w", w.s.kind, w.table, err)
	}
	w.written += uint64(w.pendingRows)
	w.pending = w.pending[:0]
	w.pendingRows = 0
	if w.written >= w.commitEvery {
		if err := w.commit(); err != nil {
			return err
		}
		w.written = 0
	}
	return nil
}

// commit 收掉当前事务并开新的 (长事务会撑爆 undo/日志, 也拖慢 MySQL)。
func (w *sqlWriter) commit() error {
	for _, s := range w.stmts {
		_ = s.Close()
	}
	w.stmts = map[int]*sql.Stmt{}
	if err := w.tx.Commit(); err != nil {
		return fmt.Errorf("提交 %s 表 %s: %w", w.s.kind, w.table, err)
	}
	tx, err := w.s.db.Begin()
	if err != nil {
		return fmt.Errorf("重新开事务: %w", err)
	}
	w.tx = tx
	return nil
}

func (w *sqlWriter) Close() error {
	if w.closed {
		return nil
	}
	w.closed = true
	if err := w.flush(); err != nil {
		_ = w.tx.Rollback()
		return err
	}
	for _, s := range w.stmts {
		_ = s.Close()
	}
	if err := w.tx.Commit(); err != nil {
		return fmt.Errorf("提交 %s 表 %s: %w", w.s.kind, w.table, err)
	}
	return nil
}

// Discard 放弃尚未提交的内容 (migrate.Discarder, 出错路径用)。
//
// 与 Close 的区别只有收尾那一步: Close 会把缓冲区 flush 掉再 Commit, Discard 直接
// Rollback 并丢掉缓冲。已经跨过 commitEvery 阈值提交过的批次留在库里 —— 那是分批
// 提交的既有语义 (长事务会撑爆 undo/日志), 能挽回的是当前事务和缓冲区里这一批;
// 报告里的 Rows 会如实说明写到哪儿了。
func (w *sqlWriter) Discard() error {
	if w.closed {
		return nil
	}
	w.closed = true
	for _, s := range w.stmts {
		_ = s.Close()
	}
	w.stmts = map[int]*sql.Stmt{}
	w.pending = w.pending[:0]
	w.pendingRows = 0
	if err := w.tx.Rollback(); err != nil && !errors.Is(err, sql.ErrTxDone) {
		return fmt.Errorf("回滚 %s 表 %s: %w", w.s.kind, w.table, err)
	}
	return nil
}

// ============================ 保真表 ============================

// metaDTO 保真表里一行 (JSON) 的内容。字段名特意取短: 这张表会跟着数据一起躺在
// 别人的库里, 越不占地方越好; 版本号留着以后改格式时兜底。
type metaDTO struct {
	Version     int       `json:"v"`
	PK          string    `json:"pk,omitempty"`
	PKIndexName string    `json:"pkn,omitempty"`
	PKIndexType byte      `json:"pkt,omitempty"`
	RowStore    []string  `json:"rs,omitempty"`
	Columns     []metaCol `json:"cols"`
	Indexes     []metaIdx `json:"idx,omitempty"`
}

type metaCol struct {
	Name string `json:"n"`
	Type byte   `json:"t"`
	// Param 列级类型参数 (目前是 Decimal 的 scale)。SQL 的 DECIMAL(38,2) 声明虽然
	// 带着小数位, 但 SQLite 的 TEXT 列看不出来 —— 存进保真表才能原样还原。
	Param byte `json:"p,omitempty"`
	// Time 源 SQL 的时间语义 ("datetime" / "date" / "timestamp"), 空即普通列。
	// 新迁的表盘上它是 FieldType_DateTime (Unix 毫秒), 靠这个标记导回 SQL 时才还原成
	// 当初的 DATETIME / DATE / TIMESTAMP —— 光看 FieldType 只能分出"是时间"。
	// 旧库可能是 Uint64+纳秒 (更早的实现), 读路径按列的 FieldType 分岔兼容, 两条路都在。
	Time string `json:"tm,omitempty"`
	// Null 该列可空。SQL 侧的可空性虽然能从 PRAGMA / information_schema 读出来,
	// 但那是【建表时写下的声明】; 我们导出时按 VRTX 的 NULL 位图写 NULL / NOT NULL,
	// 带一份在保真表里, 回来就不必依赖各家 SQL 方言的可空推断。
	Null bool `json:"nl,omitempty"`
	// Uniq 该列声明了唯一约束。列位与索引唯一位是两件事 (声明 vs 约束),
	// 各存各的: 只存索引的话, 导回 VRTX 后列上的声明就没了。
	Uniq bool `json:"uq,omitempty"`
}

type metaIdx struct {
	Name    string   `json:"n"`
	Type    byte     `json:"t"`
	Columns []string `json:"c"`
	// Uniq 唯一索引 (SQL 侧的 UNIQUE INDEX / VRTX 的 IndexFlagUnique)。
	Uniq bool `json:"u,omitempty"`
}

const metaVersion = 1

func (s *SQL) ensureMetaTable() error {
	if _, err := s.db.Exec(s.dial.metaDDL()); err != nil {
		return fmt.Errorf("建保真表 %s: %w", MetaTable, err)
	}
	return nil
}

func (s *SQL) writeMeta(sc *TableSchema) error {
	if err := s.ensureMetaTable(); err != nil {
		return err
	}
	dto := metaDTO{
		Version:     metaVersion,
		PK:          sc.PKColumn,
		PKIndexName: sc.PKIndexName,
		PKIndexType: byte(sc.PKIndexType),
		RowStore:    sc.RowStoreColumns,
	}
	for _, c := range sc.Columns {
		dto.Columns = append(dto.Columns, metaCol{
			Name: c.Name, Type: byte(c.Type), Param: c.Param, Time: c.TimeKind,
			Null: c.Nullable, Uniq: c.Unique,
		})
	}
	for _, idx := range sc.Indexes {
		dto.Indexes = append(dto.Indexes, metaIdx{
			Name: idx.Name, Type: byte(idx.Type), Columns: idx.Columns, Uniq: idx.Unique,
		})
	}
	payload, err := json.Marshal(dto)
	if err != nil {
		return err
	}
	if _, err := s.db.Exec(s.dial.metaUpsert(), sc.Name, string(payload)); err != nil {
		return fmt.Errorf("写保真表 %s: %w", MetaTable, err)
	}
	return nil
}

func (s *SQL) readMeta(table string) (*metaDTO, bool) {
	if s.hasMeta == nil {
		s.hasMeta = map[string]*metaDTO{}
	}
	if m, ok := s.hasMeta[table]; ok {
		return m, m != nil
	}
	// 保真表不存在是常态 (外部库), 静默跳过
	if ok, err := s.Exists(MetaTable); err != nil || !ok {
		s.hasMeta[table] = nil
		return nil, false
	}
	var payload string
	err := s.db.QueryRow(s.dial.metaSelect(), table).Scan(&payload)
	if err != nil {
		if !errors.Is(err, sql.ErrNoRows) {
			s.opts.logf("[%s] 读保真表 %s 失败 (按 SQL 类型推断): %v", s.kind, MetaTable, err)
		}
		s.hasMeta[table] = nil
		return nil, false
	}
	var dto metaDTO
	if err := json.Unmarshal([]byte(payload), &dto); err != nil {
		s.opts.logf("[%s] 保真表里表 %s 的内容解析失败 (按 SQL 类型推断): %v", s.kind, table, err)
		s.hasMeta[table] = nil
		return nil, false
	}
	s.hasMeta[table] = &dto
	return &dto, true
}

// applyMeta 用保真数据覆盖推断出来的结构。
//
// 只覆盖"双方都有"的列: 目标 SQL 表里后来手工加的列照常按 SQL 类型推断,
// 不会因为保真表里没有它就丢列。
func applyMeta(s *TableSchema, m *metaDTO) {
	typeOf := make(map[string]types.FieldType, len(m.Columns))
	timeOf := make(map[string]string, len(m.Columns))
	paramOf := make(map[string]byte, len(m.Columns))
	nullOf := make(map[string]bool, len(m.Columns))
	uniqOf := make(map[string]bool, len(m.Columns))
	for _, c := range m.Columns {
		typeOf[c.Name] = types.FieldType(c.Type)
		if c.Param != 0 {
			paramOf[c.Name] = c.Param
		}
		if c.Time != "" {
			timeOf[c.Name] = c.Time
		}
		nullOf[c.Name] = c.Null
		uniqOf[c.Name] = c.Uniq
	}
	for i := range s.Columns {
		if t, ok := typeOf[s.Columns[i].Name]; ok {
			s.Columns[i].Type = t
			// NULL / 唯一声明无条件取自保真表: 旧格式 (没这两个字段) 解出来是 false,
			// 而旧格式导出的表当年就是"全列 NOT NULL + 无唯一约束", 语义正好对上。
			s.Columns[i].Nullable = nullOf[s.Columns[i].Name]
			s.Columns[i].Unique = uniqOf[s.Columns[i].Name]
		}
		// Param 无条件覆盖: 保真表里没记 (旧格式) 时是 0, 而推断出来的 SQL DECIMAL
		// 也可能带 scale —— 以保真表为准才叫"还原当初的结构"。
		if p, ok := paramOf[s.Columns[i].Name]; ok {
			s.Columns[i].Param = p
		} else if _, ok := typeOf[s.Columns[i].Name]; ok {
			s.Columns[i].Param = 0
		}
		if k, ok := timeOf[s.Columns[i].Name]; ok {
			s.Columns[i].TimeKind = k
		}
	}
	if m.PK != "" && s.HasColumn(m.PK) {
		s.PKColumn = m.PK
	}
	if len(m.RowStore) > 0 {
		s.RowStoreColumns = append([]string(nil), m.RowStore...)
	}
	if m.PKIndexName != "" {
		s.PKIndexName = m.PKIndexName
	}
	if m.PKIndexType != 0 {
		s.PKIndexType = page.IndexType(m.PKIndexType)
	}
	// 索引: 按名字合并, 保真表里的类型更可信 (SQL 侧根本区分不出 Hash/BTree)
	byName := make(map[string]metaIdx, len(m.Indexes))
	for _, idx := range m.Indexes {
		byName[idx.Name] = idx
	}
	for i := range s.Indexes {
		if mi, ok := byName[s.Indexes[i].Name]; ok {
			s.Indexes[i].Type = page.IndexType(mi.Type)
			s.Indexes[i].Unique = mi.Uniq
			if len(mi.Columns) > 0 {
				s.Indexes[i].Columns = mi.Columns
			}
			delete(byName, mi.Name)
		}
	}
	for _, mi := range m.Indexes {
		if _, ok := byName[mi.Name]; !ok {
			continue
		}
		if isRowStoreType(page.IndexType(mi.Type)) {
			continue
		}
		s.Indexes = append(s.Indexes, IndexSchema{
			Name:    mi.Name,
			Type:    page.IndexType(mi.Type),
			Columns: mi.Columns,
			Unique:  mi.Uniq,
		})
	}
}

// isRowStoreType 主键/时序这类"承载行数据"的索引类型。
func isRowStoreType(t page.IndexType) bool {
	return t == page.IndexTypePrimary || t == page.IndexTypeTime
}