package migrate

// TEXT 列采样定宽的用例: 真实 SQLite 源 → VRTX, 验证采样结果落成了哪一档、
// 索引有没有跟着保住, 以及关掉开关后的旧行为。风格与 roundtrip_test.go 一致。

import (
	"git.ycl.pub/yl32940/TsDB2/DB/core"
	"git.ycl.pub/yl32940/TsDB2/DB/core/types"
	"git.ycl.pub/yl32940/TsDB2/DB/core/util"
	"database/sql"
	"fmt"
	"path/filepath"
	"strings"
	"testing"

	_ "modernc.org/sqlite"
)

// vrtxColumn 直接打开 VRTX 文件读一列的引擎定义 —— 与 migrate 的读端无关,
// 宽度码按 Flag 低 5 位读 (与 vrtxFields 落盘的方式一致)。
func vrtxColumn(t *testing.T, path, table, col string) types.FieldInfo {
	t.Helper()
	db, err := core.Open(path)
	if err != nil {
		t.Fatalf("打开 %s: %v", path, err)
	}
	defer db.Close()
	tbl, ok := db.FindTable(table)
	if !ok {
		t.Fatalf("%s 里没有表 %s", path, table)
	}
	for _, f := range tbl.GetFields() {
		if util.BytesToString(f.FieldName) == col {
			return f
		}
	}
	t.Fatalf("表 %s 里没有列 %s", table, col)
	return types.FieldInfo{}
}

// colWidth 列的行内宽度 (字节): 宽度码在 Flag 低 5 位。
func colWidth(f types.FieldInfo) int {
	return int(types.FieldSize(f.FieldType, f.Flag&types.FlagScaleMask))
}

// buildSQLiteTexts 建一张带 TEXT 列的源表:
//
//	body TEXT  最大 100 字节 (ASCII), 上面有索引 —— 采到 128 档后索引必须建出来
//	memo TEXT  最大 5 字节, 含 NULL —— 采到 32 档
//	tag  TEXT  20 个汉字 = 60 字节 —— 钉住"采样数的是字节不是字符" (SQLite 的
//	         length() 对文本数字符, 不 CAST 成 BLOB 会采成 20 → 32 档, 写入即超宽)
func buildSQLiteTexts(t *testing.T, path string) {
	t.Helper()
	db, err := sql.Open("sqlite", path)
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	for _, s := range []string{
		`CREATE TABLE Docs (
			id   INTEGER PRIMARY KEY,
			body TEXT,
			memo TEXT,
			tag  TEXT
		)`,
		`CREATE INDEX idx_Docs_body ON Docs(body)`,
	} {
		if _, err := db.Exec(s); err != nil {
			t.Fatalf("执行 %s: %v", s, err)
		}
	}
	rows := []struct {
		id   int
		body string
		memo any
		tag  string
	}{
		{1, strings.Repeat("x", 100), "short", strings.Repeat("汉", 20)}, // 60 字节
		{2, "hi", nil, ""},
	}
	for _, r := range rows {
		if _, err := db.Exec(`INSERT INTO Docs (id, body, memo, tag) VALUES (?, ?, ?, ?)`,
			r.id, r.body, r.memo, r.tag); err != nil {
			t.Fatalf("插入 %v: %v", r, err)
		}
	}
}

// TestTextWidthScanToVRTX TEXT 列按源端采样落成定长档: 100 字节 → 128 档、短列 → 32 档、
// 多字节文本按【字节】采样 (60 字节 → 64 档); 源表建在 TEXT 列上的索引因此不再被跳过。
func TestTextWidthScanToVRTX(t *testing.T) {
	dir := t.TempDir()
	srcPath := filepath.Join(dir, "src.sqlite")
	dstPath := filepath.Join(dir, "dst.db")
	buildSQLiteTexts(t, srcPath)

	src, err := OpenSQLite(srcPath, quiet())
	if err != nil {
		t.Fatal(err)
	}
	dst, err := OpenVRTX(dstPath, quiet())
	if err != nil {
		t.Fatal(err)
	}
	rep, err := Run(src, dst, quiet())
	if err != nil {
		t.Fatalf("SQLite→VRTX: %v", err)
	}
	_ = src.Close()
	_ = dst.Close()

	if rep.Rows != 2 {
		t.Fatalf("应迁 2 行, 实际 %d", rep.Rows)
	}

	// 三列都应是定长 string, 档位精确等于 2 次幂 (不是向上漂到 224 之类):
	// 采样 100 → 128, 采样 5 → 32, 采样 60 (字节!) → 64。
	for col, wantWidth := range map[string]int{"body": 128, "memo": 32, "tag": 64} {
		f := vrtxColumn(t, dstPath, "Docs", col)
		if f.FieldType != types.FieldType_String {
			t.Fatalf("列 %s 应是 FieldType_String, 实际 %v (定宽档位没生效)", col, f.FieldType)
		}
		if got := colWidth(f); got != wantWidth {
			t.Errorf("列 %s 的定宽应为 %d 字节, 实际 %d (Flag 低 5 位 = 宽度码 %d)",
				col, wantWidth, got, f.Flag&types.FlagScaleMask)
		}
		if code := f.Flag & types.FlagScaleMask; code != types.WidthCodeFor(wantWidth) {
			t.Errorf("列 %s 的宽度码 %d 应是 WidthCodeFor(%d) = %d",
				col, code, wantWidth, types.WidthCodeFor(wantWidth))
		}
	}

	// 建在 TEXT 列上的索引: 定宽后必须真的建出来 (旧行为里它因"动态列不能当索引键"被跳过)。
	snap := snapshotVRTX(t, dstPath, "Docs")
	if got := snap.indexes["idx_Docs_body"]; got != "btree(body)" {
		t.Errorf("TEXT 列上的索引应随采样定宽建出来, 实际 %v", snap.indexes)
	}

	// 报告: 采样结果要写进 Note (定了哪些档), 且不该再有"跳过索引"。
	if len(rep.Tables) != 1 {
		t.Fatalf("应有 1 张表的报告, 实际 %d", len(rep.Tables))
	}
	note := rep.Tables[0].Note
	for _, want := range []string{"采样定宽", "body 128 字节", "memo 32 字节", "tag 64 字节"} {
		if !strings.Contains(note, want) {
			t.Errorf("报告 Note 里缺少 %q: %q", want, note)
		}
	}
	if strings.Contains(note, "跳过索引") {
		t.Errorf("定宽后不该再跳过索引: %q", note)
	}
}

// TestTextWidthScanNullAndOversize 两个边界: 全 NULL 的 TEXT 列采到 0, 落最窄的 32 档;
// 采样超过 2048 字节的列保持动态列 (并写进报告)。
func TestTextWidthScanNullAndOversize(t *testing.T) {
	dir := t.TempDir()
	srcPath := filepath.Join(dir, "src.sqlite")
	dstPath := filepath.Join(dir, "dst.db")

	db, err := sql.Open("sqlite", srcPath)
	if err != nil {
		t.Fatal(err)
	}
	for _, s := range []string{
		`CREATE TABLE Notes (
			id    INTEGER PRIMARY KEY,
			empty TEXT,
			big   TEXT
		)`,
		`INSERT INTO Notes (id, empty, big) VALUES (1, NULL, ?)`,
	} {
		if _, err := db.Exec(s, strings.Repeat("y", 5000)); err != nil {
			t.Fatalf("执行 %s: %v", s, err)
		}
	}
	_ = db.Close()

	src, err := OpenSQLite(srcPath, quiet())
	if err != nil {
		t.Fatal(err)
	}
	dst, err := OpenVRTX(dstPath, quiet())
	if err != nil {
		t.Fatal(err)
	}
	rep, err := Run(src, dst, quiet())
	if err != nil {
		t.Fatalf("SQLite→VRTX: %v", err)
	}
	_ = src.Close()
	_ = dst.Close()

	// 全 NULL → 0 → 32 档。
	empty := vrtxColumn(t, dstPath, "Notes", "empty")
	if empty.FieldType != types.FieldType_String || colWidth(empty) != 32 {
		t.Errorf("全 NULL 的 TEXT 列应落 32 档, 实际 %v 宽 %d", empty.FieldType, colWidth(empty))
	}

	// 5000 字节 > 2048: 保持动态列。
	big := vrtxColumn(t, dstPath, "Notes", "big")
	if big.FieldType != types.FieldType_Dynamicbyte {
		t.Errorf("采样超 2048 字节的列应保持动态列, 实际 %v", big.FieldType)
	}
	if len(rep.Tables) != 1 {
		t.Fatalf("应有 1 张表的报告, 实际 %d", len(rep.Tables))
	}
	note := rep.Tables[0].Note
	for _, want := range []string{"采样定宽", "empty 32 字节", "保持动态", "big"} {
		if !strings.Contains(note, want) {
			t.Errorf("报告 Note 里缺少 %q: %q", want, note)
		}
	}
	// 数据照常迁过去 (5000 字节的值装得进动态列)。
	snap := snapshotVRTX(t, dstPath, "Notes")
	if len(snap.rows) != 1 {
		t.Errorf("应迁 1 行, 实际 %d", len(snap.rows))
	}
}

// TestTextWidthScanDisabled TextWidthScan=false 退回旧行为: TEXT 家族全落动态列,
// 上面的索引照旧被跳过并写进报告。
func TestTextWidthScanDisabled(t *testing.T) {
	dir := t.TempDir()
	srcPath := filepath.Join(dir, "src.sqlite")
	dstPath := filepath.Join(dir, "dst.db")
	buildSQLiteTexts(t, srcPath)

	src, err := OpenSQLite(srcPath, quiet())
	if err != nil {
		t.Fatal(err)
	}
	dst, err := OpenVRTX(dstPath, quiet())
	if err != nil {
		t.Fatal(err)
	}
	o := quiet()
	o.TextWidthScan = false
	rep, err := Run(src, dst, o)
	if err != nil {
		t.Fatalf("SQLite→VRTX: %v", err)
	}
	_ = src.Close()
	_ = dst.Close()

	for _, col := range []string{"body", "memo", "tag"} {
		f := vrtxColumn(t, dstPath, "Docs", col)
		if f.FieldType != types.FieldType_Dynamicbyte {
			t.Errorf("关掉采样后列 %s 应落动态列 (旧行为), 实际 %v", col, f.FieldType)
		}
	}

	snap := snapshotVRTX(t, dstPath, "Docs")
	if _, ok := snap.indexes["idx_Docs_body"]; ok {
		t.Errorf("动态列上的索引不该建出来: %v", snap.indexes)
	}
	if len(rep.Tables) != 1 {
		t.Fatalf("应有 1 张表的报告, 实际 %d", len(rep.Tables))
	}
	note := rep.Tables[0].Note
	if !strings.Contains(note, "跳过索引") || !strings.Contains(note, "idx_Docs_body") {
		t.Errorf("动态列的索引跳过必须写进报告: %q", note)
	}
	if strings.Contains(note, "采样定宽") {
		t.Errorf("关掉采样后不该再有采样定宽的记录: %q", note)
	}
}

// TestTextWidthCode 宽度规则本身: 大于等于 maxLen 的最小 2 次幂, 下限 32, 上限 2048;
// 2048 < maxLen ≤ MaxStringSize 保持动态列; maxLen > MaxStringSize 落 Blob 链式列。
// 档位必须精确命中 (WidthCodeWidth 就是 2 次幂本身, 不是 224 之类)。
func TestTextWidthCode(t *testing.T) {
	cases := []struct {
		maxLen    int64
		wantWidth int
		wantClass textWidthClass
	}{
		{0, 32, widthToFixed},   // 全 NULL / 空表
		{1, 32, widthToFixed},   // 下限 32: 任何非空值至少 32 档
		{16, 32, widthToFixed},  // 16 字节也抬到 32 (下限)
		{32, 32, widthToFixed},
		{33, 64, widthToFixed},
		{100, 128, widthToFixed},
		{2048, 2048, widthToFixed},
		{2049, 0, widthToDynamic}, // 超定宽上限: 保持动态列
		{5000, 0, widthToDynamic},
		{16360, 0, widthToDynamic}, // 正好顶到单值上限: 动态列仍装得下
		{16361, 0, widthToBlob},    // 超单值上限: 落 Blob 链式列
		{60000, 0, widthToBlob},
		{1 << 30, 0, widthToBlob},
	}
	for _, tc := range cases {
		code, width, class := textWidthCode(tc.maxLen)
		if class != tc.wantClass || width != tc.wantWidth {
			t.Errorf("textWidthCode(%d) = (码 %d, 宽 %d, 态 %d), 期望 (宽 %d, 态 %d)",
				tc.maxLen, code, width, class, tc.wantWidth, tc.wantClass)
			continue
		}
		if class == widthToFixed && int(types.WidthCodeWidth(code)) != width {
			t.Errorf("textWidthCode(%d): 宽度码 %d 对应 %d 字节, 与档位 %d 不符 (没有精确命中)",
				tc.maxLen, code, types.WidthCodeWidth(code), width)
		}
	}
}

// TestIsTextFamilyDecl 只有 TEXT 家族参与采样: JSON/BLOB/BINARY 这类同样落动态列的
// 类型必须被排除, 带长度的 CHAR/VARCHAR 走声明容量取档, 也不参与。
func TestIsTextFamilyDecl(t *testing.T) {
	for _, decl := range []string{
		"TEXT", "text", " Text ", "TINYTEXT", "MEDIUMTEXT", "LONGTEXT", "CLOB", "NCLOB",
		"TEXT(65535)", // MySQL information_schema 会给 TEXT 报一个类型自身上界
	} {
		if !isTextFamilyDecl(decl) {
			t.Errorf("%q 应判定为 TEXT 家族", decl)
		}
	}
	for _, decl := range []string{
		"", "JSON", "JSONB", "BLOB", "TINYBLOB", "LONGBLOB", "BINARY", "VARBINARY",
		"VARCHAR(32)", "CHAR(8)", "NVARCHAR(50)", "DATETIME", "DECIMAL(18,2)",
	} {
		if isTextFamilyDecl(decl) {
			t.Errorf("%q 不应判定为 TEXT 家族", decl)
		}
	}
}

// TestTextWidthsOneQueryPerTable 同表多列一次查询取回 (TextWidthSampler 的约定),
// 且全 NULL 列采到 0 —— 直接钉 SQL.TextWidths, 不经过 Run。
func TestTextWidthsOneQueryPerTable(t *testing.T) {
	dir := t.TempDir()
	srcPath := filepath.Join(dir, "src.sqlite")
	db, err := sql.Open("sqlite", srcPath)
	if err != nil {
		t.Fatal(err)
	}
	for _, s := range []string{
		`CREATE TABLE T (id INTEGER PRIMARY KEY, a TEXT, b TEXT)`,
		`INSERT INTO T (id, a, b) VALUES (1, '12345', NULL)`,
	} {
		if _, err := db.Exec(s); err != nil {
			t.Fatalf("执行 %s: %v", s, err)
		}
	}
	_ = db.Close()

	sqlSrc, err := OpenSQLite(srcPath, quiet())
	if err != nil {
		t.Fatal(err)
	}
	defer sqlSrc.Close()
	var src Source = sqlSrc // 走 Source 接口形态, 钉住运行期可选能力断言
	sampler, ok := src.(TextWidthSampler)
	if !ok {
		t.Fatal("*SQL 应实现 TextWidthSampler")
	}
	got, err := sampler.TextWidths("T", []string{"a", "b"})
	if err != nil {
		t.Fatalf("采样: %v", err)
	}
	if got["a"] != 5 || got["b"] != 0 {
		t.Errorf("采样结果应为 a=5, b=0 (全 NULL), 实际 %v", got)
	}
}


// TestTextWidthOversizeFallsToBlob 采样超过动态列单值上限 (types.MaxStringSize, 16360)
// 的 TEXT 列不再"保持动态列、写入时被引擎按单块上限拒掉整表", 而是落成 Blob 链式列:
// 60KB 的值逐字节迁进 VRTX (经引擎 BlobRead 对照), 再逐字节迁回 SQLite —— 复刻
// DockerManager 实测的失败场景 (sys_logs.param 59566 字节)。
func TestTextWidthOversizeFallsToBlob(t *testing.T) {
	dir := t.TempDir()
	srcPath := filepath.Join(dir, "src.sqlite")
	midPath := filepath.Join(dir, "mid.db")
	backPath := filepath.Join(dir, "back.sqlite")

	big := strings.Repeat("0123456789abcdef", 3750) // 60000 字节, 超 16360 上限

	db, err := sql.Open("sqlite", srcPath)
	if err != nil {
		t.Fatal(err)
	}
	for _, s := range []string{
		`CREATE TABLE sys_logs (
			id     INTEGER PRIMARY KEY,
			action VARCHAR(32),
			param  TEXT
		)`,
		`INSERT INTO sys_logs (id, action, param) VALUES (1, 'deploy', ?)`,
	} {
		if _, err := db.Exec(s, big); err != nil {
			t.Fatalf("执行 %s: %v", s, err)
		}
	}
	// 同一列再补一行短值 + 一行 NULL: 落 Blob 后小值与 NULL 的读写也必须正常。
	if _, err := db.Exec(`INSERT INTO sys_logs (id, action, param) VALUES (2, 'ping', 'ok')`); err != nil {
		t.Fatal(err)
	}
	if _, err := db.Exec(`INSERT INTO sys_logs (id, action, param) VALUES (3, 'noop', NULL)`); err != nil {
		t.Fatal(err)
	}
	_ = db.Close()

	src, err := OpenSQLite(srcPath, quiet())
	if err != nil {
		t.Fatal(err)
	}
	mid, err := OpenVRTX(midPath, quiet())
	if err != nil {
		t.Fatal(err)
	}
	rep, err := Run(src, mid, quiet())
	if err != nil {
		t.Fatalf("SQLite→VRTX (60KB TEXT 应自动落 Blob 迁过去): %v", err)
	}
	_ = src.Close()
	_ = mid.Close()

	// 列类型: 必须是 Blob 链式列, 而不是保持动态列等写入时失败。
	f := vrtxColumn(t, midPath, "sys_logs", "param")
	if f.FieldType != types.FieldType_Blob {
		t.Fatalf("采样超单值上限的列 param 应落 FieldType_Blob, 实际 %v", f.FieldType)
	}

	// 内容逐字节对照 (不经 migrate 读端, 直接开引擎文件读链页, 免得"两端同一段代码"
	// 把不对称的错抹平): 快照 norm 用既有 bytes:%x 形态。
	vdb, err := core.Open(midPath)
	if err != nil {
		t.Fatal(err)
	}
	vtbl, ok := vdb.FindTable("sys_logs")
	if !ok {
		t.Fatal("mid.db 里没有 sys_logs")
	}
	got := map[uint64]string{}
	err = vtbl.Query().ScanWithPKID(func(pkID uint64, vals []types.FieldValue) bool {
		// NULL / 空 blob 的行内引用读侧一律"读不到" (位图置位 / 空引用跳过),
		// 缺该列即 NULL; 先按 NULL 预置, 见到值再覆盖。
		got[pkID] = "NULL"
		for _, fv := range vals {
			if util.BytesToString(fv.FieldName) != "param" {
				continue
			}
			ref, ok := fv.Value.(types.BlobRef)
			if !ok {
				t.Fatalf("param 的行内值应是 BlobRef, 实际 %T", fv.Value)
			}
			data, err := vtbl.BlobRead(ref, types.CompressNone)
			if err != nil {
				t.Fatalf("读回 param 的 blob 链: %v", err)
			}
			got[pkID] = fmt.Sprintf("bytes:%x", data)
		}
		return true
	})
	_ = vdb.Close()
	if err != nil {
		t.Fatal(err)
	}
	wantBig := fmt.Sprintf("bytes:%x", []byte(big))
	if got[1] != wantBig {
		t.Errorf("60KB 值迁过去后内容变了: 长度 got=%d want=%d", len(got[1]), len(wantBig))
	}
	if got[2] != fmt.Sprintf("bytes:%x", []byte("ok")) {
		t.Errorf("短值行 param = %q", got[2])
	}
	if got[3] != "NULL" {
		t.Errorf("NULL 行 param = %q, 想要 NULL", got[3])
	}

	// 报告: 落 Blob 必须可见。
	if len(rep.Tables) != 1 {
		t.Fatalf("应有 1 张表的报告, 实际 %d", len(rep.Tables))
	}
	note := rep.Tables[0].Note
	for _, w := range []string{"落 Blob 链式列", "60000 字节", "param"} {
		if !strings.Contains(note, w) {
			t.Errorf("报告 Note 里缺少 %q: %q", w, note)
		}
	}

	// 反向再迁一轮 (VRTX → SQLite): Blob 列经 migrate 读端解引用、按 BLOB 列建表,
	// 内容必须逐字节回来。
	vmid, err := OpenVRTX(midPath, quiet())
	if err != nil {
		t.Fatal(err)
	}
	back, err := OpenSQLite(backPath, quiet())
	if err != nil {
		t.Fatal(err)
	}
	if _, err := Run(vmid, back, quiet()); err != nil {
		t.Fatalf("VRTX→SQLite: %v", err)
	}
	_ = vmid.Close()
	_ = back.Close()

	bdb, err := sql.Open("sqlite", backPath)
	if err != nil {
		t.Fatal(err)
	}
	defer bdb.Close()
	var backBig, backShort []byte
	var backNull sql.NullString
	if err := bdb.QueryRow(`SELECT param FROM sys_logs WHERE id = 1`).Scan(&backBig); err != nil {
		t.Fatal(err)
	}
	if fmt.Sprintf("bytes:%x", backBig) != wantBig {
		t.Errorf("迁回 SQLite 的 60KB 值逐字节不一致 (got %d 字节)", len(backBig))
	}
	if err := bdb.QueryRow(`SELECT param FROM sys_logs WHERE id = 2`).Scan(&backShort); err != nil {
		t.Fatal(err)
	}
	if string(backShort) != "ok" {
		t.Errorf("短值行迁回 = %q", backShort)
	}
	if err := bdb.QueryRow(`SELECT param FROM sys_logs WHERE id = 3`).Scan(&backNull); err != nil {
		t.Fatal(err)
	}
	if backNull.Valid {
		t.Errorf("NULL 行迁回应是 NULL, 实际 %q", backNull.String)
	}
}

// TestCharDeclOversizeUpgrade 声明长度的字符列"实际数据超档"的采样升档 —— 复刻
// DockerManager 实测: SQLite 的 sys_ssl.cert 声明 VARCHAR(2048), 但 SQLite 不强制
// 声明长度, 实际值 4803 字节; 旧行为按声明建 2048 定长档, 写入即报"值 4803 字节
// 超过定长列容量 2048"。现在这类列与 TEXT 列合一句采样, 按 textWidthCode 三态升档:
// cert (声明 2048 / 实际 4803) → 动态列; payload (声明 8192 / 实际 9000) → 动态列;
// bulky (声明 8192 / 实际 20000, 超动态列单值上限) → Blob 链式列; code (声明 100 /
// 实际 100) → 定长 128 档不动。数据逐字节往返, 报告写升档 note。
func TestCharDeclOversizeUpgrade(t *testing.T) {
	dir := t.TempDir()
	srcPath := filepath.Join(dir, "src.sqlite")
	midPath := filepath.Join(dir, "mid.db")
	backPath := filepath.Join(dir, "back.sqlite")

	cert := strings.Repeat("c", 4803)
	payload := strings.Repeat("m", 9000)
	bulky := strings.Repeat("h", 20000)

	db, err := sql.Open("sqlite", srcPath)
	if err != nil {
		t.Fatal(err)
	}
	if _, err := db.Exec(`CREATE TABLE sys_ssl (
		id      INTEGER PRIMARY KEY,
		cert    VARCHAR(2048),
		payload VARCHAR(8192),
		bulky   VARCHAR(8192),
		code    VARCHAR(100)
	)`); err != nil {
		t.Fatal(err)
	}
	if _, err := db.Exec(`INSERT INTO sys_ssl (id, cert, payload, bulky, code) VALUES (1, ?, ?, ?, ?)`,
		cert, payload, bulky, strings.Repeat("k", 100)); err != nil {
		t.Fatal(err)
	}
	// 第二行短值 + NULL: 升档后小值与 NULL 的读写也必须正常。
	if _, err := db.Exec(`INSERT INTO sys_ssl (id, cert, payload, bulky, code) VALUES (2, 'ok', NULL, NULL, 'go')`); err != nil {
		t.Fatal(err)
	}
	_ = db.Close()

	src, err := OpenSQLite(srcPath, quiet())
	if err != nil {
		t.Fatal(err)
	}
	mid, err := OpenVRTX(midPath, quiet())
	if err != nil {
		t.Fatal(err)
	}
	rep, err := Run(src, mid, quiet())
	if err != nil {
		t.Fatalf("SQLite→VRTX (超档字符列应自动升档迁过去): %v", err)
	}
	_ = src.Close()
	_ = mid.Close()

	if rep.Rows != 2 {
		t.Fatalf("应迁 2 行, 实际 %d", rep.Rows)
	}

	// 列类型: cert / payload 升 (保持) 动态列, bulky 超单值上限落 Blob, code 定长不动。
	if f := vrtxColumn(t, midPath, "sys_ssl", "cert"); f.FieldType != types.FieldType_Dynamicbyte {
		t.Errorf("列 cert (声明 2048 / 实际 4803) 应升档动态列, 实际 %v", f.FieldType)
	}
	if f := vrtxColumn(t, midPath, "sys_ssl", "payload"); f.FieldType != types.FieldType_Dynamicbyte {
		t.Errorf("列 payload (声明 8192 / 实际 9000) 应为动态列, 实际 %v", f.FieldType)
	}
	if f := vrtxColumn(t, midPath, "sys_ssl", "bulky"); f.FieldType != types.FieldType_Blob {
		t.Errorf("列 bulky (声明 8192 / 实际 20000, 超单值上限) 应落 Blob 链式列, 实际 %v", f.FieldType)
	}
	if f := vrtxColumn(t, midPath, "sys_ssl", "code"); f.FieldType != types.FieldType_String || colWidth(f) != 128 {
		t.Errorf("列 code (声明 100 / 实际 100) 应保持定长 128 档, 实际 %v 宽 %d", f.FieldType, colWidth(f))
	}

	// 值逐字节对照 (不经 migrate 读端, 直接开引擎文件读, 免得"两端同一段代码"把
	// 不对称的错抹平): 定长/动态列读回 []byte, Blob 列是 BlobRef 要走 BlobRead;
	// NULL 列读侧缺项 (位图置位跳过), 缺列即 NULL。
	vdb, err := core.Open(midPath)
	if err != nil {
		t.Fatal(err)
	}
	vtbl, ok := vdb.FindTable("sys_ssl")
	if !ok {
		t.Fatal("mid.db 里没有 sys_ssl")
	}
	got := map[uint64]map[string]string{}
	err = vtbl.Query().ScanWithPKID(func(pkID uint64, vals []types.FieldValue) bool {
		m := map[string]string{}
		for _, fv := range vals {
			name := util.BytesToString(fv.FieldName)
			switch name {
			case "cert", "payload", "bulky", "code":
			default:
				continue // id (行号列) 等其它列不在逐字节对照范围
			}
			if ref, isBlob := fv.Value.(types.BlobRef); isBlob {
				data, err := vtbl.BlobRead(ref, types.CompressNone)
				if err != nil {
					t.Fatalf("读回 %s 的 blob 链: %v", name, err)
				}
				m[name] = fmt.Sprintf("bytes:%x", data)
				continue
			}
			b, ok := fv.Value.([]byte)
			if !ok {
				t.Fatalf("列 %s 的行内值应是 []byte, 实际 %T", name, fv.Value)
			}
			m[name] = fmt.Sprintf("bytes:%x", b)
		}
		got[pkID] = m
		return true
	})
	_ = vdb.Close()
	if err != nil {
		t.Fatal(err)
	}
	wantRow1 := map[string]string{
		"cert":    fmt.Sprintf("bytes:%x", []byte(cert)),
		"payload": fmt.Sprintf("bytes:%x", []byte(payload)),
		"bulky":   fmt.Sprintf("bytes:%x", []byte(bulky)),
		"code":    fmt.Sprintf("bytes:%x", []byte(strings.Repeat("k", 100))),
	}
	for col, want := range wantRow1 {
		if got[1][col] != want {
			t.Errorf("行 1 列 %s 逐字节不一致 (got %d 字节)", col, len(got[1][col]))
		}
	}
	if got[2]["cert"] != fmt.Sprintf("bytes:%x", []byte("ok")) ||
		got[2]["code"] != fmt.Sprintf("bytes:%x", []byte("go")) {
		t.Errorf("行 2 短值行 = %v", got[2])
	}
	if _, ok := got[2]["payload"]; ok {
		t.Errorf("行 2 的 payload 应是 NULL, 实际有值")
	}
	if _, ok := got[2]["bulky"]; ok {
		t.Errorf("行 2 的 bulky 应是 NULL, 实际有值")
	}

	// 报告: 升档必须可见, 且只有 cert / bulky 升档 (payload 没超单值上限,
	// code 没超档宽, 都不该出现在升档名单里)。
	if len(rep.Tables) != 1 {
		t.Fatalf("应有 1 张表的报告, 实际 %d", len(rep.Tables))
	}
	note := rep.Tables[0].Note
	for _, want := range []string{
		"升档",
		"列 cert 声明 2048 字节但实际最大 4803 字节, 升档 动态列",
		"列 bulky 声明 8192 字节但实际最大 20000 字节, 升档 Blob 链式列",
	} {
		if !strings.Contains(note, want) {
			t.Errorf("报告 Note 里缺少 %q: %q", want, note)
		}
	}
	if strings.Contains(note, "payload 声明") {
		t.Errorf("payload (9000 ≤ 单值上限) 不该升档: %q", note)
	}
	if strings.Contains(note, "code 声明") {
		t.Errorf("code (数据没超档) 不该升档: %q", note)
	}

	// 反向再迁一轮 (VRTX → SQLite): 升档后的列经 migrate 读端解出来建表,
	// 内容必须逐字节回来。
	vmid, err := OpenVRTX(midPath, quiet())
	if err != nil {
		t.Fatal(err)
	}
	back, err := OpenSQLite(backPath, quiet())
	if err != nil {
		t.Fatal(err)
	}
	if _, err := Run(vmid, back, quiet()); err != nil {
		t.Fatalf("VRTX→SQLite: %v", err)
	}
	_ = vmid.Close()
	_ = back.Close()

	bdb, err := sql.Open("sqlite", backPath)
	if err != nil {
		t.Fatal(err)
	}
	defer bdb.Close()
	var backCert, backPayload, backBulky, backCode []byte
	if err := bdb.QueryRow(`SELECT cert, payload, bulky, code FROM sys_ssl WHERE id = 1`).
		Scan(&backCert, &backPayload, &backBulky, &backCode); err != nil {
		t.Fatal(err)
	}
	if fmt.Sprintf("bytes:%x", backCert) != wantRow1["cert"] {
		t.Errorf("迁回 SQLite 的 cert 逐字节不一致 (got %d 字节)", len(backCert))
	}
	if fmt.Sprintf("bytes:%x", backPayload) != wantRow1["payload"] {
		t.Errorf("迁回 SQLite 的 payload 逐字节不一致 (got %d 字节)", len(backPayload))
	}
	if fmt.Sprintf("bytes:%x", backBulky) != wantRow1["bulky"] {
		t.Errorf("迁回 SQLite 的 bulky 逐字节不一致 (got %d 字节)", len(backBulky))
	}
	if fmt.Sprintf("bytes:%x", backCode) != wantRow1["code"] {
		t.Errorf("迁回 SQLite 的 code 逐字节不一致 (got %d 字节)", len(backCode))
	}
	var backShort string
	var backNull1, backNull2 sql.NullString
	if err := bdb.QueryRow(`SELECT cert, payload, bulky FROM sys_ssl WHERE id = 2`).
		Scan(&backShort, &backNull1, &backNull2); err != nil {
		t.Fatal(err)
	}
	if backShort != "ok" {
		t.Errorf("短值行迁回 cert = %q", backShort)
	}
	if backNull1.Valid || backNull2.Valid {
		t.Errorf("NULL 行迁回应是 NULL, 实际 %q / %q", backNull1.String, backNull2.String)
	}
}

// TestDeclaredCharWidth 只有带长度的 CHAR/VARCHAR 家族声明返回声明字节数 —— 这是
// 字符列采样升档那一路的入场券; TEXT 家族 / 无长度声明 / 非字符类型都返回 0,
// 不采样。名单与 inferSQLType 的字符列分支同源。
func TestDeclaredCharWidth(t *testing.T) {
	for decl, want := range map[string]int{
		"VARCHAR(2048)": 2048, // DockerManager 实测的 sys_ssl.cert
		"varchar(100)":  100,
		"CHAR(8)":       8,
		"CHARACTER(16)": 16,
		"NCHAR(4)":      4,
		"NVARCHAR(50)":  50,
		"VARCHAR2(32)":  32,
		" VARCHAR(64) ": 64,
		"VARCHAR(8192)": 8192, // 声明超 2048 落成动态列, 仍要采样 (超单值上限落 Blob)
	} {
		if got := declaredCharWidth(decl); got != want {
			t.Errorf("declaredCharWidth(%q) = %d, 期望 %d", decl, got, want)
		}
	}
	for _, decl := range []string{
		"", "TEXT", "TINYTEXT", "CLOB", "BLOB", "JSON", "BINARY", "VARBINARY",
		"VARCHAR", "CHAR", "NCHAR", "DECIMAL(18,2)", "DATETIME", "INT",
	} {
		if got := declaredCharWidth(decl); got != 0 {
			t.Errorf("declaredCharWidth(%q) = %d, 期望 0 (不参与字符列采样)", decl, got)
		}
	}
}