TsDB2/migrate/textwidth_test.go

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"
	"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 不可定宽。档位必须精确命中 (WidthCodeWidth 就是 2 次幂本身, 不是 224 之类)。
func TestTextWidthCode(t *testing.T) {
	cases := []struct {
		maxLen    int64
		wantWidth int
		wantOK    bool
	}{
		{0, 32, true},   // 全 NULL / 空表
		{1, 32, true},   // 下限 32: 任何非空值至少 32 档
		{16, 32, true},  // 16 字节也抬到 32 (下限)
		{32, 32, true},
		{33, 64, true},
		{100, 128, true},
		{2048, 2048, true},
		{2049, 0, false}, // 超上限: 保持动态列
		{5000, 0, false},
	}
	for _, tc := range cases {
		code, width, ok := textWidthCode(tc.maxLen)
		if ok != tc.wantOK || width != tc.wantWidth {
			t.Errorf("textWidthCode(%d) = (码 %d, 宽 %d, ok %v), 期望 (宽 %d, ok %v)",
				tc.maxLen, code, width, ok, tc.wantWidth, tc.wantOK)
			continue
		}
		if ok && 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)
	}
}