TsDB2/migrate/value.go

package migrate

import (
	"git.ycl.pub/yl32940/TsDB2/DB/core/types"
	"fmt"
	"math"
	"math/big"
	"strconv"
	"strings"
	"time"
	"unicode/utf8"
)

// ============================ 值转换 ============================
//
// 两端拿到的值形态不同:
//   VRTX 侧 (deserializeRow 的产出): uint8 / int16 / int32 / uint64 / float64 / []byte
//   SQL 侧 (database/sql 的驱动值):  int64 / float64 / []byte / string / nil / time.Time
//
// 中间表示不做统一 (会把值搬两遍), 而是让目标端按自己的列类型把来源值"收编":
// 走到 VRTX 用 coerceForVRTX, 走到 SQL 用 coerceForSQL。

// zeroForVRTX 某种 VRTX 列类型的零值。
//
// VRTX 没有 NULL 这个概念 (变长列的行内引用 PageID==0 就是"空"), 所以 SQL 的 NULL
// 落到这里只能是零值 —— 这是单向的信息损失, 文档里明说。
func zeroForVRTX(ft types.FieldType) any {
	switch ft {
	case types.FieldType_Uint8:
		return uint8(0)
	case types.FieldType_Uint16:
		return int16(0)
	case types.FieldType_Uint32:
		return int32(0)
	case types.FieldType_Uint64:
		return uint64(0)
	case types.FieldType_DateTime:
		// 引擎 DateTime 列的零值: 零毫秒 (Unix 0)。非空列上的"没有时间"就落它。
		return uint64(0)
	case types.FieldType_Int8:
		return int8(0)
	case types.FieldType_Int16:
		return int16(0)
	case types.FieldType_Int32:
		return int32(0)
	case types.FieldType_Int64:
		return int64(0)
	case types.FieldType_Float64:
		return float64(0)
	case types.FieldType_Float32:
		return float32(0)
	case types.FieldType_Bool:
		return false
	case types.FieldType_Decimal:
		return types.Decimal{}
	case types.FieldType_UUID:
		return types.UUID{}
	case types.FieldType_Year:
		return int16(0)
	case types.FieldType_Int128:
		return new(big.Int)
	default:
		return []byte{}
	}
}

// coerceForVRTX 把来源值规范成 core.writeFieldValue 认得的确切 Go 类型。
//
// 必须逐一对齐 (uint8→uint8, uint16→int16, uint32→int32, uint64→uint64,
// Int8→int8, Int16→int16, Int32→int32, Int64→int64,
// float64→float64, 定长/变长→[]byte): writeFieldValue 内部是类型断言,
// 给错类型不是"写进去一个错值", 而是直接 panic。
//
// param 是列级参数 (Decimal 的 scale): 定点列必须按它对齐小数位, 否则同一列里
// 混进另一种 scale 的行, B+ 树的字节序比较会静默出错。
//
// nullable 决定 nil 怎么落地: 可空列写 NULL (原样把 nil 交给 core, 由它置 NULL
// 位图的位), 不可空列才退化成零值。
func coerceForVRTX(ft types.FieldType, param byte, v any, col string, strict bool, nullable bool) (any, error) {
	if v == nil {
		// SQL 侧的 NULL 读出来就是 nil (见 SQL.Scan 的 *any)。可空列必须把它写成
		// NULL —— 落成零值等于"把没有值"变成"值是 0 / 空串", 而且不可逆: 导回去
		// 分不出这行的 0 原本是 NULL 还是真 0。
		if nullable {
			return nil, nil
		}
		return zeroForVRTX(ft), nil
	}
	switch ft {
	case types.FieldType_Uint8:
		n, err := asInt64(v)
		if err != nil {
			return nil, fmt.Errorf("列 %s: %w", col, err)
		}
		if n < 0 || n > math.MaxUint8 {
			if strict {
				return nil, fmt.Errorf("列 %s (uint8): 值 %d 超出 0..255", col, n)
			}
			n &= 0xFF
		}
		return uint8(n), nil

	case types.FieldType_Uint16:
		n, err := asInt64(v)
		if err != nil {
			return nil, fmt.Errorf("列 %s: %w", col, err)
		}
		// 无符号档盘上存的是 16 位位模式, 读取载体是 int16 (见 codec.DecodeValue 与
		// writeFieldValue 里的 value.(int16))。所以"装得下"的判据是这 16 位放得下,
		// 而不是"int16 的正数区放得下":
		//	0..65535      无符号来源的原样值 (MySQL SMALLINT UNSIGNED 的 33400 就在这一档);
		//	-32768..-1    回读 / SQLite INTEGER 给出的负位模式 (0x8278 读出来是 -32136)。
		// 按 int16 区间判会把 32768..65535 这半个无符号区误判成越界: 位模式明明装得下,
		// 报错拦掉的却是完全合法的数据。
		if strict && (n < math.MinInt16 || n > math.MaxUint16) {
			return nil, fmt.Errorf("列 %s (uint16): 值 %d 超出 16 位位模式可表达的 -32768..65535", col, n)
		}
		return int16(uint16(n)), nil

	case types.FieldType_Uint32:
		n, err := asInt64(v)
		if err != nil {
			return nil, fmt.Errorf("列 %s: %w", col, err)
		}
		// 同 uint16: 32 位位模式, 载体 int32。0..4294967295 是无符号来源的原样值,
		// -2147483648..-1 是回读给出的负位模式。
		if strict && (n < math.MinInt32 || n > math.MaxUint32) {
			return nil, fmt.Errorf("列 %s (uint32): 值 %d 超出 32 位位模式可表达的 -2147483648..4294967295", col, n)
		}
		return int32(uint32(n)), nil

	case types.FieldType_DateTime:
		// 引擎专用时间列 (Unix 毫秒): time.Time 直传、文本解析、整数按毫秒 ——
		// 换算约定见 coerceDateTimeForVRTX。
		return coerceDateTimeForVRTX(v, col, nullable)

	case types.FieldType_Uint64:
		// 有符号来源按"64 位位模式"原样搬: int64(-1) → uint64(0xFFFF...)。
		// 导出回 SQL 时再用 int64(u) 写, 于是 MySQL 的 -1 ↔ VRTX 的 0xFFFF...
		// 往返无损 —— VRTX 的 uint64 本身不分正负, 同一个位模式 Go 侧既可当
		// uint64 也可当 int64 读, 正是 ORM 让 int64 字段能存负数的原理。
		u, err := asUint64(v)
		if err != nil {
			return nil, fmt.Errorf("列 %s: %w", col, err)
		}
		return u, nil

	case types.FieldType_Int8:
		n, err := asInt64(v)
		if err != nil {
			return nil, fmt.Errorf("列 %s: %w", col, err)
		}
		if strict && (n < math.MinInt8 || n > math.MaxInt8) {
			return nil, fmt.Errorf("列 %s (int8): 值 %d 超出 -128..127", col, n)
		}
		return int8(n), nil

	case types.FieldType_Int16:
		n, err := asInt64(v)
		if err != nil {
			return nil, fmt.Errorf("列 %s: %w", col, err)
		}
		if strict && (n < math.MinInt16 || n > math.MaxInt16) {
			return nil, fmt.Errorf("列 %s (int16): 值 %d 超出 int16", col, n)
		}
		return int16(n), nil

	case types.FieldType_Int32:
		n, err := asInt64(v)
		if err != nil {
			return nil, fmt.Errorf("列 %s: %w", col, err)
		}
		if strict && (n < math.MinInt32 || n > math.MaxInt32) {
			return nil, fmt.Errorf("列 %s (int32): 值 %d 超出 int32", col, n)
		}
		return int32(n), nil

	case types.FieldType_Int64:
		n, err := asInt64(v)
		if err != nil {
			return nil, fmt.Errorf("列 %s: %w", col, err)
		}
		return n, nil

	case types.FieldType_Float64:
		f, err := asFloat64(v)
		if err != nil {
			return nil, fmt.Errorf("列 %s: %w", col, err)
		}
		return f, nil

	// ===== 扩展类型: 一律归到 core 的标准解码载体 =====
	// (Bool→bool, Float32→float32, Decimal→types.Decimal, UUID→types.UUID,
	//  Year→int16, Int128→*big.Int) —— 与 core/types.DecodeValue 反着走一遍。

	case types.FieldType_Float32:
		return coerceFloat32(v, col, strict)

	case types.FieldType_Bool:
		n, err := asInt64(v)
		if err != nil {
			return nil, fmt.Errorf("列 %s: %w", col, err)
		}
		if strict && n != 0 && n != 1 {
			return nil, fmt.Errorf("列 %s (bool): 值 %d 不是 0/1 —— "+
				"bool 只认 0/1, 否则等值与索引会给出多套答案", col, n)
		}
		return n != 0, nil

	case types.FieldType_Decimal:
		d, err := types.DecimalFromValue(v)
		if err != nil {
			return nil, fmt.Errorf("列 %s: %w", col, err)
		}
		if uint16(param) != d.Scale() {
			d, err = d.Rescale(uint16(param))
			if err != nil {
				return nil, fmt.Errorf("列 %s: %w", col, err)
			}
		}
		return d, nil

	case types.FieldType_UUID:
		u, err := asUUID(v)
		if err != nil {
			return nil, fmt.Errorf("列 %s: %w", col, err)
		}
		return u, nil

	case types.FieldType_Year:
		n, err := asInt64(v)
		if err != nil {
			return nil, fmt.Errorf("列 %s: %w", col, err)
		}
		if strict && (n < 0 || n > math.MaxInt16) {
			return nil, fmt.Errorf("列 %s (year): 值 %d 超出 0..32767", col, n)
		}
		return int16(n), nil

	case types.FieldType_Int128:
		bi, err := asBigInt(v)
		if err != nil {
			return nil, fmt.Errorf("列 %s: %w", col, err)
		}
		return bi, nil

	case types.FieldType_Blob:
		// Blob 列: 中间表示里给的是【内容本身】(VRTX.Scan 已把行内的 8B 引用解成字节),
		// 这里原样交给 core —— 行内那 8 字节链引用由 core 的 Table.blobFieldRef 落链页得到
		// (见 types/blob.go: "把 io.Reader / []byte 写进 FieldType_Blob 列, 引擎自动走链页")。
		//
		// 刻意不走下面 default 的定长容量闸门: types.FieldSize(Blob) 只有 8 —— 那是行内引用的
		// 宽度, 不是容量。拿它当上限会把任何超过 8 字节的 blob 判成"超过定长列容量"。单值上限
		// 由链长决定 (约 4GB), 超了 core 会返 ErrBlobTooLarge。
		// 也不补零 / 不裁零: blob 是二进制, 0x00 是内容的一部分。
		b, err := asBytes(v)
		if err != nil {
			return nil, fmt.Errorf("列 %s: %w", col, err)
		}
		return b, nil
	default:
		// 定长字符串 / 变长列
		b, err := asBytes(v)
		if err != nil {
			return nil, fmt.Errorf("列 %s: %w", col, err)
		}
		if !types.IsDynamic(ft) {
			// 容量取列的真实宽度 (宽度码), 不是类型默认档: 对声明 1024 字节的列按 256
			// 判断会把合法值判成超长。
			size := int(types.FieldSize(ft, param))
			if len(b) > size {
				if strict {
					return nil, fmt.Errorf("列 %s (%s): 值 %d 字节超过定长列容量 %d 字节, "+
						"导入会截断 —— 请把该列改成 dynamic 或放宽源值", col, TypeName(ft), len(b), size)
				}
				b = b[:size]
			}
		}
		return b, nil
	}
}

// coerceFloat32 收编到单精度。来源已是 float32 时原样; float64 只在 strict 下要求
// 无损 —— 让"迁移会丢精度"这件事在严格档下当场暴露, 而不是事后对不上数。
func coerceFloat32(v any, col string, strict bool) (float32, error) {
	if f32, ok := v.(float32); ok {
		return f32, nil
	}
	f, err := asFloat64(v)
	if err != nil {
		return 0, fmt.Errorf("列 %s: %w", col, err)
	}
	fb := float32(f)
	if strict && float64(fb) != f {
		return 0, fmt.Errorf("列 %s (float32): 值 %v 用 float32 表示会损失精度", col, f)
	}
	return fb, nil
}

// asUUID 把来源值收成 UUID: UUID 值 / 16 字节 / 文本 (带或不带连字符)。
func asUUID(v any) (types.UUID, error) {
	var u types.UUID
	switch x := v.(type) {
	case types.UUID:
		return x, nil
	case []byte:
		if len(x) != types.UUIDSize {
			return u, fmt.Errorf("UUID 需要 %d 字节, 得到 %d 字节", types.UUIDSize, len(x))
		}
		copy(u[:], x)
		return u, nil
	case string:
		return types.ParseUUID(x)
	default:
		return u, fmt.Errorf("无法把 %T 当 UUID", v)
	}
}

// asBigInt 把来源值收成 *big.Int (128 位整数的标准载体)。
func asBigInt(v any) (*big.Int, error) {
	switch x := v.(type) {
	case *big.Int:
		if x == nil {
			return new(big.Int), nil
		}
		return new(big.Int).Set(x), nil
	case big.Int:
		return new(big.Int).Set(&x), nil
	case types.Decimal:
		if x.Scale() != 0 {
			return nil, fmt.Errorf("int128 列不能接收带小数的值 %s", x.String())
		}
		return x.Coefficient(), nil
	case []byte:
		return parseInt128(string(x))
	case string:
		return parseInt128(x)
	case uint64:
		return new(big.Int).SetUint64(x), nil
	case int64:
		return big.NewInt(x), nil
	default:
		n, err := asInt64(v)
		if err != nil {
			return nil, err
		}
		return big.NewInt(n), nil
	}
}

// parseInt128 解析十进制/十六进制文本的 128 位整数 (base 0 支持 0x 前缀)。
func parseInt128(s string) (*big.Int, error) {
	s = strings.TrimSpace(s)
	bi, ok := new(big.Int).SetString(s, 0)
	if !ok {
		return nil, fmt.Errorf("无法把 %q 当 128 位整数", s)
	}
	return bi, nil
}

// coerceForSQL 把来源值规范成 database/sql 能绑定的类型。
//
// 三个约定:
//   - uint64 必须按方言选 Go 类型 (mysql 参数), 因为两个方言的声明本就不同,
//     绑定类型得跟着走 —— 见下面 FieldType_Uint64 分支。
//   - 字节类列 (定长串 / 变长) 若是合法 UTF-8 就绑成 string, 否则绑 []byte。
//     前者让 SQLite 把它存成 TEXT (BLOB 亲和性的列不会自动转换),
//     后者保证二进制列不被改坏。定长列导出时去掉尾部补零。
//   - 时间列一律还原成时间文本, 不写整数: FieldType_DateTime 列按【毫秒】解释,
//     存量 Uint64+TimeKind 旧表按【纳秒】解释 (两条路的判别就是列的 FieldType)。
//
// mysql 来自目标方言 (Sink), 与 sqlTypeName 的第二个参数同源。
func coerceForSQL(c Column, v any, mysql bool) (any, error) {
	if v == nil {
		// 可空列: 原样写 NULL。以前所有列都不可空, 这里一路把 nil 落成零值 ——
		// 在可空列上那是【静默改数据】: VRTX 的 NULL 到了 SQL 变成 0 / 空串 /
		// 1970-01-01, 而且不可逆, 再导回来分不出它原本是 NULL。
		if c.Nullable {
			return nil, nil
		}
		if c.IsDynamic() {
			return []byte{}, nil
		}
		if c.IsTime() {
			// 时间列的零值给 Unix 0, 不是数字 0 —— 后者塞进 DATETIME 列不是
			// "1970" 而是让 MySQL 报错 (严格模式) 或存成 0000-00-00。
			return formatSQLTime(0, c.TimeKind), nil
		}
		return zeroForSQL(c), nil
	}
	switch c.Type {
	case types.FieldType_Uint8:
		n, err := asInt64(v)
		if err != nil {
			return nil, fmt.Errorf("列 %s: %w", c.Name, err)
		}
		return int64(uint8(n)), nil

	case types.FieldType_Uint16:
		n, err := asInt64(v)
		if err != nil {
			return nil, fmt.Errorf("列 %s: %w", c.Name, err)
		}
		// 出库一侧给的是【无符号数值】, 不是 int64 原样 —— 与 uint8 那条同形。
		// 位模式 0x8278 读出来是 -32136, 原样绑 int64 送进 MySQL 的 SMALLINT UNSIGNED
		// (sqlTypeName 就是这么声明的) 会被判成 Out of range: 严格模式直接报错, 非严格
		// 模式截成 0 —— 那是静默改数据。取回无符号值 33400 才是这一列真实存的东西。
		// SQLite 侧声明是 INTEGER, 没有无符号概念, 但 0..65535 它整个装得下, 同样给
		// 无符号值即可 (只有 uint64 才必须按位模式降级, 因为 SQLite 的 INTEGER 装不下
		// 2^63 以上)。
		return int64(uint16(n)), nil

	case types.FieldType_Uint32:
		n, err := asInt64(v)
		if err != nil {
			return nil, fmt.Errorf("列 %s: %w", c.Name, err)
		}
		// 同 uint16: 给无符号数值, 别把 4294967295 当成 -1 送给 MySQL 的 INT UNSIGNED。
		return int64(uint32(n)), nil

	case types.FieldType_DateTime:
		// 时间列: 盘上 (引擎 DecodeValue) 是 uint64 毫秒, 回 SQL 得还原成时间文本 ——
		// 否则 MySQL 的 DATETIME 列会收到一个 1.7e12 的整数 (驱动不会替我们转)。
		// time.Time 先按绝对时间取毫秒, 其余形态按数值 (毫秒) 收。
		var ms int64
		switch x := v.(type) {
		case time.Time:
			ms = x.UnixMilli()
		default:
			u, err := asUint64(v)
			if err != nil {
				return nil, fmt.Errorf("列 %s: %w", c.Name, err)
			}
			ms = int64(u)
		}
		return formatSQLTime(ms, c.TimeKind), nil

	case types.FieldType_Uint64:
		u, err := asUint64(v)
		if err != nil {
			return nil, fmt.Errorf("列 %s: %w", c.Name, err)
		}
		if c.IsTime() {
			// 存量旧表: 这列是当年迁成 uint64 存 Unix【纳秒】的时间列 (meta 里带
			// TimeKind 标记), 回 SQL 仍按纳秒还原成时间文本。新表是 FieldType_DateTime,
			// 走上面那条毫秒路 —— 两条路并存, 判别依据就是列的 FieldType。
			return formatSQLTimeNS(int64(u), c.TimeKind), nil
		}
		// 绑定的 Go 类型必须跟目标列的声明一致, 而两个方言的声明不同:
		//   - MySQL: sqlTypeName 声明成 BIGINT UNSIGNED (uint64 是无符号档),
		//     所以绑 uint64 —— 整个 2^64 区间都写得进去。以前一律绑 int64(u),
		//     高位为 1 的值 (包括 VRTX 从 int64 负值搬过来的位模式) 在严格模式下
		//     会被 MySQL 判成 "Out of range value" 直接报错, 非严格模式则被静默改写。
		//   - SQLite: 只有有符号的 INTEGER, 绑 int64 —— 位模式原样存进 64 位整数,
		//     读回来还是同一个 8 字节。
		// 两条路共同的保证是【位模式不变】, 这正是 VRTX uint64 的语义 (不分正负)。
		if mysql {
			return u, nil
		}
		return int64(u), nil

	case types.FieldType_Int8, types.FieldType_Int16, types.FieldType_Int32, types.FieldType_Int64:
		// 有符号档存的就是数值本身 (盘上的保序编码由 core 负责), 直接绑 int64
		n, err := asInt64(v)
		if err != nil {
			return nil, fmt.Errorf("列 %s: %w", c.Name, err)
		}
		return n, nil

	case types.FieldType_Float64:
		f, err := asFloat64(v)
		if err != nil {
			return nil, fmt.Errorf("列 %s: %w", c.Name, err)
		}
		return f, nil

	// ===== 扩展类型 =====
	// Decimal / Int128 一律绑文本: 目标列分别是 DECIMAL(38,s)/VARCHAR(40) (MySQL) 与
	// TEXT (SQLite), 文本既能精确表达 128 位整数, 也不会被驱动转成浮点。
	// UUID 绑 16 字节 (BINARY(16)/BLOB)。Bool 绑 bool, Year 绑 int64。

	case types.FieldType_Float32:
		f, err := asFloat64(v)
		if err != nil {
			return nil, fmt.Errorf("列 %s: %w", c.Name, err)
		}
		return float32(f), nil

	case types.FieldType_Bool:
		n, err := asInt64(v)
		if err != nil {
			return nil, fmt.Errorf("列 %s: %w", c.Name, err)
		}
		return n != 0, nil

	case types.FieldType_Decimal:
		d, err := types.DecimalFromValue(v)
		if err != nil {
			return nil, fmt.Errorf("列 %s: %w", c.Name, err)
		}
		if uint16(c.Param) != d.Scale() {
			d, err = d.Rescale(uint16(c.Param))
			if err != nil {
				return nil, fmt.Errorf("列 %s: %w", c.Name, err)
			}
		}
		return d.String(), nil

	case types.FieldType_UUID:
		u, err := asUUID(v)
		if err != nil {
			return nil, fmt.Errorf("列 %s: %w", c.Name, err)
		}
		return append([]byte(nil), u[:]...), nil

	case types.FieldType_Year:
		n, err := asInt64(v)
		if err != nil {
			return nil, fmt.Errorf("列 %s: %w", c.Name, err)
		}
		return n, nil

	case types.FieldType_Int128:
		bi, err := asBigInt(v)
		if err != nil {
			return nil, fmt.Errorf("列 %s: %w", c.Name, err)
		}
		return bi.String(), nil

	case types.FieldType_Blob:
		// Blob 列: 中间表示里已是内容本身 (VRTX.Scan 解过引用), 原样绑给 SQL。
		//
		// 不做 UTF-8 判定也不去尾零 —— 那是"字符串列"的规矩 (见 default 分支)。blob 是二进制,
		// 把末尾的 0x00 当补位裁掉就是静默改数据; 绑成 string 还会让 SQLite 按 TEXT 亲和性存,
		// 真正的二进制或非 UTF-8 内容会被写坏。
		b, err := asBytes(v)
		if err != nil {
			return nil, fmt.Errorf("列 %s: %w", c.Name, err)
		}
		return b, nil
	default:
		b, err := asBytes(v)
		if err != nil {
			return nil, fmt.Errorf("列 %s: %w", c.Name, err)
		}
		if !c.IsDynamic() {
			b = trimTrailingNUL(b)
		}
		if utf8.Valid(b) {
			return string(b), nil
		}
		return b, nil
	}
}

func zeroForSQL(c Column) any {
	switch c.Type {
	case types.FieldType_Float64:
		return float64(0)
	case types.FieldType_Float32:
		return float32(0)
	case types.FieldType_Bool:
		return false
	case types.FieldType_Decimal:
		// 零值也要按列 scale 输出定点文本 ("0.00"), 否则 DECIMAL 列会收到 "0"。
		return types.NewDecimalInt64(0, uint16(c.Param)).String()
	case types.FieldType_UUID:
		return make([]byte, types.UUIDSize)
	case types.FieldType_Year:
		return int64(0)
	case types.FieldType_Int128:
		return "0"
	case types.FieldType_Blob:
		// blob 列的零值是"空内容"(长度 0), 不是行内那 8 字节链引用 —— core 收到空 []byte
		// 会给出空引用 (不分配链页)。
		return []byte{}
	case types.FieldType_Uint8, types.FieldType_Uint16, types.FieldType_Uint32, types.FieldType_Uint64,
		types.FieldType_Int8, types.FieldType_Int16, types.FieldType_Int32, types.FieldType_Int64:
		return int64(0)
	case types.FieldType_DateTime:
		// 与 coerceForSQL 的 nil 分支同一约定: 零时间给 Unix 0 (1970-01-01), 不给数字 0。
		return formatSQLTime(0, "datetime")
	default:
		return ""
	}
}

// trimTrailingNUL 去掉定长列右补的 \x00。
// VRTX 定长字符串是"右补零到固定宽度", SQL 侧不该带上这些补位 (否则 varchar
// 里会多出一堆看不见的字符, 也没法用等值条件查)。
func trimTrailingNUL(b []byte) []byte {
	end := len(b)
	for end > 0 && b[end-1] == 0 {
		end--
	}
	return b[:end]
}

// ============================ 底层取值 ============================

func asInt64(v any) (int64, error) {
	switch x := v.(type) {
	case int64:
		return x, nil
	case int:
		return int64(x), nil
	case int32:
		return int64(x), nil
	case int16:
		return int64(x), nil
	case int8:
		return int64(x), nil
	case uint64:
		if x > math.MaxInt64 {
			return 0, fmt.Errorf("值 %d 超出 int64", x)
		}
		return int64(x), nil
	case uint32:
		return int64(x), nil
	case uint16:
		return int64(x), nil
	case uint8:
		return int64(x), nil
	case uint:
		return int64(x), nil
	case float64:
		return roundFloatToInt(x)
	case float32:
		return roundFloatToInt(float64(x))
	case bool:
		if x {
			return 1, nil
		}
		return 0, nil
	case []byte:
		return parseIntOrTime(string(x))
	case string:
		return parseIntOrTime(x)
	case time.Time:
		return x.UnixNano(), nil
	default:
		return 0, fmt.Errorf("无法把 %T 当整数", v)
	}
}

func roundFloatToInt(f float64) (int64, error) {
	if math.IsNaN(f) || math.IsInf(f, 0) {
		return 0, fmt.Errorf("值 %v 不是有限数", f)
	}
	// 上界必须是 >= 而不是 >: math.MaxInt64 (2^63-1) 在 float64 里存不下, 会被舍入
	// 到 2^63 —— 于是 f == 2^63 时 "f > MaxInt64" 是假, 放过去之后 int64(f) 越界,
	// Go 对越界的 float→int 转换是未定义结果 (amd64 上给 MinInt64), 静默存成反数。
	// 下界不受这个影响: MinInt64 = -2^63 在 float64 里可以精确表示。
	if f >= math.MaxInt64 || f < math.MinInt64 {
		return 0, fmt.Errorf("值 %v 超出 int64", f)
	}
	return int64(math.Round(f)), nil
}

// asUint64 与 asInt64 的差别只有一处: 有符号的负值不报错, 而是按位模式搬进 uint64。
func asUint64(v any) (uint64, error) {
	switch x := v.(type) {
	case uint64:
		return x, nil
	case uint32:
		return uint64(x), nil
	case uint16:
		return uint64(x), nil
	case uint8:
		return uint64(x), nil
	case uint:
		return uint64(x), nil
	case int64:
		return uint64(x), nil // 位模式, 负数 → 高位全 1
	case int:
		return uint64(x), nil
	case int32:
		return uint64(x), nil
	case int16:
		return uint64(x), nil
	case int8:
		return uint64(x), nil
	case float64:
		n, err := roundFloatToInt(x)
		if err != nil {
			return 0, err
		}
		return uint64(n), nil
	case float32:
		n, err := roundFloatToInt(float64(x))
		if err != nil {
			return 0, err
		}
		return uint64(n), nil
	case bool:
		if x {
			return 1, nil
		}
		return 0, nil
	case []byte:
		return parseUintOrTime(string(x))
	case string:
		return parseUintOrTime(x)
	case time.Time:
		return uint64(x.UnixNano()), nil
	default:
		return 0, fmt.Errorf("无法把 %T 当整数", v)
	}
}

// ============================ 时间列 ============================
//
// 时间列在 VRTX 侧有两种盘面, 读路径按【列的 FieldType】分岔:
//
//   - FieldType_DateTime (新): 引擎专用时间类型, 行内 8 字节 Unix【毫秒】;
//   - FieldType_Uint64 + TimeKind 标记 (旧): 更早的实现把时间列迁成 uint64 存
//     Unix【纳秒】, 存量表的数据就是这个盘面 —— 判别靠 TimeKind (新表是 DateTime,
//     旧表才会出现 Uint64 列带着它)。
//
// SQL 时间列迁进来要换算成目标盘面, 迁出去要换算回时间文本 —— 这一节就是这两个方向。

// sqlTimeLayouts SQL 时间文本的常见写法。MySQL 的 DATETIME / TIMESTAMP 打出来
// 就是第一个; SQLite 存的是 TEXT, 可能带 T 和时区; DATE 只有日期部分。
var sqlTimeLayouts = []string{
	"2006-01-02 15:04:05.999999999",
	"2006-01-02 15:04:05",
	"2006-01-02T15:04:05.999999999Z07:00",
	"2006-01-02T15:04:05",
	"2006-01-02",
}

// parseSQLTime 把 SQL 时间文本解析成 time.Time。
//
// 一律按 UTC 解释 (除非文本自带偏移): MySQL 的 DATETIME 本身不带时区, 若用本地
// 时区去解释, 同一份数据在不同机器上会迁出不同的时间戳 —— 迁移工具不该引入
// 这种漂移。往返一致性优先于"看着像本地时间"。
func parseSQLTime(s string) (time.Time, bool) {
	for _, layout := range sqlTimeLayouts {
		if tm, err := time.ParseInLocation(layout, s, time.UTC); err == nil {
			return tm, true
		}
	}
	return time.Time{}, false
}

// parseIntOrTime 先按整数读, 读不动再按时间文本读 (时间文本 → Unix 纳秒)。
//
// 这是【旧盘面】的入口: 时间列映射成 uint64 后 (存量 Uint64+TimeKind 表), 值可能
// 是 time.Time (MySQL 驱动开了 parseTime), 也可能是文本 (parseTime=false 的默认档,
// 以及 SQLite 的 DATETIME 列) —— 后者必须在这里换算, 否则写进 uint64 列时会被
// 当成"不是数字"直接报错。新的 FieldType_DateTime 列不走这里 (见 coerceDateTimeForVRTX)。
func parseIntOrTime(s string) (int64, error) {
	s = strings.TrimSpace(s)
	if n, err := strconv.ParseInt(s, 10, 64); err == nil {
		return n, nil
	}
	if tm, ok := parseSQLTime(s); ok {
		return tm.UnixNano(), nil
	}
	return 0, fmt.Errorf("无法把 %q 当整数 (也不是能识别的时间文本)", s)
}

// parseUintOrTime 同 parseIntOrTime, 只是落在无符号上。
func parseUintOrTime(s string) (uint64, error) {
	s = strings.TrimSpace(s)
	if n, err := strconv.ParseUint(s, 10, 64); err == nil {
		return n, nil
	}
	if tm, ok := parseSQLTime(s); ok {
		return uint64(tm.UnixNano()), nil
	}
	return 0, fmt.Errorf("无法把 %q 当整数 (也不是能识别的时间文本)", s)
}

// coerceDateTimeForVRTX SQL 侧时间值 → FieldType_DateTime 列的形态 (time.Time / 毫秒数)。
//
// 引擎 codec 认两种输入: time.Time (编码层自行截毫秒) 与毫秒数值。SQL 侧的驱动值
// 是 time.Time / 文本 / 整数, 这里统一收编:
//
//   - time.Time 原样直传 (与 gormvrtx / SQL driver 的写入路径同一约定);
//   - 文本按 UTC 解析成 time.Time (parseSQLTime, 不用本地时区 —— 同一批数据在
//     不同时区的机器上必须迁出同一个时间戳);
//   - 整数按【毫秒】解释 —— DateTime 列的盘上单位就是毫秒 (Uint64+TimeKind 的
//     旧库才是纳秒, 那条路在 coerceForSQL 的 Uint64 分支)。
//
// 零时间对齐 gormvrtx 语义: 可空列写 NULL (零 time.Time 与 nil 一样表示"没有时间"),
// 非空列落零毫秒 (Unix 0)。
func coerceDateTimeForVRTX(v any, col string, nullable bool) (any, error) {
	switch x := v.(type) {
	case time.Time:
		if x.IsZero() {
			if nullable {
				return nil, nil
			}
			return uint64(0), nil
		}
		return x, nil
	case string:
		return parseSQLTimeValue(x, col)
	case []byte:
		return parseSQLTimeValue(string(x), col)
	default:
		n, err := asInt64(v)
		if err != nil {
			return nil, fmt.Errorf("列 %s: %w", col, err)
		}
		if n < 0 {
			return nil, fmt.Errorf("列 %s: 时间戳不能为负: %d", col, n)
		}
		return uint64(n), nil // 已是毫秒
	}
}

// parseSQLTimeValue 文本 → time.Time, 解析失败报出列名。
func parseSQLTimeValue(s, col string) (any, error) {
	tm, ok := parseSQLTime(strings.TrimSpace(s))
	if !ok {
		return nil, fmt.Errorf("列 %s: 无法把 %q 当时间文本", col, s)
	}
	return tm, nil
}

// formatSQLTime Unix 毫秒 → SQL 时间文本 (FieldType_DateTime 列的读路径)。
//
// DATE 只留日期; DATETIME / TIMESTAMP 留到毫秒 —— 这正是 DateTime 列的全部精度,
// 转文本无损。
func formatSQLTime(ms int64, kind string) string {
	tm := time.UnixMilli(ms).UTC()
	if kind == "date" {
		return tm.Format("2006-01-02")
	}
	if ms%1000 == 0 {
		return tm.Format("2006-01-02 15:04:05") // 整秒别拖一串 .000
	}
	return tm.Format("2006-01-02 15:04:05.000")
}

// formatSQLTimeNS Unix 纳秒 → SQL 时间文本 (存量 Uint64+TimeKind 旧表的读路径)。
//
// 只能写到微秒 (MySQL DATETIME(6) 的上限) —— 纳秒进、微秒出是旧盘面自身的精度
// 上限, 不是工具丢的。新表走 formatSQLTime, 毫秒无损。
func formatSQLTimeNS(ns int64, kind string) string {
	tm := time.Unix(0, ns).UTC()
	if kind == "date" {
		return tm.Format("2006-01-02")
	}
	if ns%int64(time.Second) == 0 {
		return tm.Format("2006-01-02 15:04:05") // 整秒别拖一串 .000000
	}
	return tm.Format("2006-01-02 15:04:05.000000")
}

func asFloat64(v any) (float64, error) {
	switch x := v.(type) {
	case float64:
		return x, nil
	case float32:
		return float64(x), nil
	case int64:
		return float64(x), nil
	case int:
		return float64(x), nil
	case int32:
		return float64(x), nil
	case int16:
		return float64(x), nil
	case int8:
		return float64(x), nil
	case uint64:
		return float64(x), nil
	case uint32:
		return float64(x), nil
	case uint16:
		return float64(x), nil
	case uint8:
		return float64(x), nil
	case []byte:
		f, err := strconv.ParseFloat(strings.TrimSpace(string(x)), 64)
		if err != nil {
			return 0, fmt.Errorf("无法把 %q 当浮点", x)
		}
		return f, nil
	case string:
		f, err := strconv.ParseFloat(strings.TrimSpace(x), 64)
		if err != nil {
			return 0, fmt.Errorf("无法把 %q 当浮点", x)
		}
		return f, nil
	default:
		return 0, fmt.Errorf("无法把 %T 当浮点", v)
	}
}

func asBytes(v any) ([]byte, error) {
	switch x := v.(type) {
	case []byte:
		return x, nil
	case string:
		return []byte(x), nil
	case nil:
		return []byte{}, nil
	case time.Time:
		return []byte(x.Format("2006-01-02 15:04:05.999999999")), nil
	case bool:
		if x {
			return []byte{1}, nil
		}
		return []byte{0}, nil
	case int64:
		return []byte(strconv.FormatInt(x, 10)), nil
	case int:
		return []byte(strconv.Itoa(x)), nil
	case int32:
		return []byte(strconv.FormatInt(int64(x), 10)), nil
	case int16:
		return []byte(strconv.FormatInt(int64(x), 10)), nil
	case int8:
		return []byte(strconv.FormatInt(int64(x), 10)), nil
	case uint64:
		return []byte(strconv.FormatUint(x, 10)), nil
	case uint32:
		return []byte(strconv.FormatUint(uint64(x), 10)), nil
	case uint16:
		return []byte(strconv.FormatUint(uint64(x), 10)), nil
	case uint8:
		return []byte(strconv.FormatUint(uint64(x), 10)), nil
	case float64:
		return []byte(strconv.FormatFloat(x, 'g', -1, 64)), nil
	case float32:
		return []byte(strconv.FormatFloat(float64(x), 'g', -1, 32)), nil
	default:
		return nil, fmt.Errorf("无法把 %T 当字节串", v)
	}
}