TsDB2/migrate/validate_test.go

package migrate

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

// ============================ 迁移侧的结构闸门 ============================
//
// 这一层存在的理由: 源库 (MySQL/SQLite) 的结构能力比 VRTX 宽, 源侧合法的东西到 VRTX 侧
// 无处安放。测的重点不只是"报错了", 而是【报错里有没有用户能据以行动的信息】——
// 源库的实际值 (900 列 / 65 字节表名) 和 VRTX 的上限 (254 / 48) 都要在里面。
// 只说一句"字段数量非法", 用户无从判断该改源表还是该拆表。

func atMostColumns(n int) []Column {
	cols := make([]Column, n)
	for i := range cols {
		cols[i] = Column{Name: "c" + strconv.Itoa(i), Type: types.FieldType_Uint8}
	}
	return cols
}

func TestValidateForVRTX(t *testing.T) {
	// ---- 列数: MySQL 允许 1017 列, VRTX 只到 core.MaxFieldsPerTable ----
	over := &TableSchema{Name: "wide", Columns: atMostColumns(core.MaxFieldsPerTable + 1)}
	err := validateForVRTX(over)
	if err == nil {
		t.Fatal("列数超限必须报错")
	}
	// 源库的事实 + 引擎的边界, 两个数都要在报错里
	for _, want := range []string{
		strconv.Itoa(core.MaxFieldsPerTable + 1), // 源表有多少列
		strconv.Itoa(core.MaxFieldsPerTable),     // VRTX 上限
		"wide",
	} {
		if !strings.Contains(err.Error(), want) {
			t.Errorf("列数超限的报错应包含 %q: %v", want, err)
		}
	}
	// 正好到上限: 放行 (闸门是 >, 不是 >=)
	exact := &TableSchema{Name: "wide", Columns: atMostColumns(core.MaxFieldsPerTable)}
	if err := validateForVRTX(exact); err != nil {
		t.Errorf("正好 %d 列应当放行: %v", core.MaxFieldsPerTable, err)
	}

	// ---- 表名: MySQL 给到 64 字节, VRTX 只到 core.TableNameSize(48) ----
	longName := strings.Repeat("t", core.TableNameSize+1)
	err = validateForVRTX(&TableSchema{Name: longName, Columns: atMostColumns(2)})
	if err == nil {
		t.Fatal("表名超长必须报错")
	}
	for _, want := range []string{
		strconv.Itoa(len(longName)),      // 源表名多长
		strconv.Itoa(core.TableNameSize), // VRTX 上限
	} {
		if !strings.Contains(err.Error(), want) {
			t.Errorf("表名超长的报错应包含 %q: %v", want, err)
		}
	}

	// ---- 列名: schema 里是 [nameLen(1B)][name...] ----
	err = validateForVRTX(&TableSchema{Name: "t", Columns: []Column{{Name: ""}}})
	if err == nil {
		t.Fatal("空列名必须报错 (会让解析侧把下一个字段的类型字节当名字)")
	}
	longCol := strings.Repeat("x", core.FieldNameSize+1)
	err = validateForVRTX(&TableSchema{Name: "t", Columns: []Column{{Name: longCol}}})
	if err == nil {
		t.Fatal("超长列名必须报错")
	}
	if !strings.Contains(err.Error(), strconv.Itoa(core.FieldNameSize)) {
		t.Errorf("超长列名的报错应给出上限 %d: %v", core.FieldNameSize, err)
	}

	// ---- 行存储索引: VRTX 只支持单列 ----
	err = validateForVRTX(&TableSchema{
		Name:            "t",
		Columns:         atMostColumns(3),
		RowStoreColumns: []string{"a", "b"},
	})
	if err == nil {
		t.Fatal("多列行存储索引必须报错")
	}
	if !strings.Contains(err.Error(), "a") || !strings.Contains(err.Error(), "b") {
		t.Errorf("行存储索引报错应点出是哪几列: %v", err)
	}

	// ---- 正常表: 一条都不该报 ----
	ok := &TableSchema{Name: "t", Columns: atMostColumns(3), PKColumn: "c0"}
	if err := validateForVRTX(ok); err != nil {
		t.Errorf("合法表不该被拦: %v", err)
	}
}

func TestAuditIndexForVRTX(t *testing.T) {
	// 索引名: SQLite 的索引名几乎没有长度限制, VRTX 只到 core.IndexNameSize
	longName := strings.Repeat("i", core.IndexNameSize+1)
	if got := auditIndexForVRTX(IndexSchema{Name: longName, Columns: []string{"a"}}); got == "" {
		t.Error("超长索引名应当被判为建不了")
	} else if !strings.Contains(got, strconv.Itoa(core.IndexNameSize)) {
		t.Errorf("超长索引名应给出上限 %d: %s", core.IndexNameSize, got)
	}

	// 分量数: IndexMeta.FieldIDs 是定长 [16]byte
	many := make([]string, core.MaxIndexFields+1)
	for i := range many {
		many[i] = "c" + strconv.Itoa(i)
	}
	got := auditIndexForVRTX(IndexSchema{Name: "idx_many", Columns: many})
	if got == "" {
		t.Fatal("分量过多的索引应当被判为建不了")
	}
	if !strings.Contains(got, strconv.Itoa(core.MaxIndexFields)) {
		t.Errorf("分量过多应给出上限 %d: %s", core.MaxIndexFields, got)
	}

	// 正常索引: 空返回 = 可以建
	if got := auditIndexForVRTX(IndexSchema{Name: "idx_ok", Columns: []string{"a", "b"}}); got != "" {
		t.Errorf("正常索引不该被判为建不了: %s", got)
	}
}

// TestDegradeNoteAndInferAgree 钉住"报的"与"做的"逐条一致。
//
// degradeNote 只负责【说】, 真正决定类型的是 inferSQLType —— 两者判定条件一旦分叉,
// 报告就成了假消息: 要么列退成了 float64 却不吭声, 要么没退却吓唬用户。所以这里对每个
// 声明同时问两边, 要求它们同进同退。
func TestDegradeNoteAndInferAgree(t *testing.T) {
	cases := []struct {
		decl string
		// wantDegrade = 期望"确实降级了"; wantWhich = 报错里该出现的关键词
		wantDegrade bool
		wantWhich   string
	}{
		{"DECIMAL(18,2)", false, ""},
		{"DECIMAL(38,2)", false, ""},   // p 到 38 仍可精确表达
		{"DECIMAL(38,31)", false, ""},  // s 到 31 仍可表达
		{"DECIMAL(39,0)", true, "39"},  // p 超 38
		{"DECIMAL(65,30)", true, "65"}, // 同上, 这是 MySQL 能建出来、VRTX 装不下的真实例子
		{"DECIMAL(38,35)", true, "35"}, // s 超 31: 系数装得下, 小数位无处表达
		{"NUMERIC(20,6)", false, ""},
		// 与 decimal 无关的声明一律不吭声 (这个函数不该对别的类型有意见)
		{"INT", false, ""},
		{"VARCHAR(4000)", false, ""},
		{"LONGTEXT", false, ""},
		{"", false, ""},
	}
	for _, c := range cases {
		note := degradeNote(c.decl)
		ft, _ := inferSQLType(c.decl, true)
		if c.wantDegrade {
			if note == "" {
				t.Errorf("%q: 退成了 float64 却没报出来 (静默丢精度)", c.decl)
				continue
			}
			if !strings.Contains(note, c.wantWhich) {
				t.Errorf("%q: 报错应指明是哪一条超限 (%s): %s", c.decl, c.wantWhich, note)
			}
			if ft != types.FieldType_Float64 {
				t.Errorf("%q: 说降级了, inferSQLType 却给了 %v", c.decl, ft)
			}
			continue
		}
		if note != "" {
			t.Errorf("%q: 没降级却报了降级: %s", c.decl, note)
		}
	}
}

// TestSQLTypeNameDecimalWidth VRTX 的 Decimal 声明出去必须装得下 VRTX 里最大的系数。
//
// 以前按 DECIMAL(MaxDecimalScale, s) = DECIMAL(31,s) 声明, 于是 scale=30 的列只剩 1 位
// 整数位, 而 VRTX 的系数上限是 10^38-1 —— 导出 8 位整数的值时会被 MySQL 判成超范围。
func TestSQLTypeNameDecimalWidth(t *testing.T) {
	for _, scale := range []byte{0, 2, 20, 30, 31} {
		decl := sqlTypeName(types.FieldType_Decimal, scale, true)
		if decl != "DECIMAL("+strconv.Itoa(types.MaxDecimalPrecision)+","+strconv.Itoa(int(scale))+")" {
			t.Errorf("Decimal(%d) 的 MySQL 声明 = %q, 应为 DECIMAL(%d,%d)",
				scale, decl, types.MaxDecimalPrecision, scale)
		}
		// 声明出去再读回来, 类型与小数位都不能变 (否则往返一次就换了语义)
		ft, param := inferSQLType(decl, true)
		if ft != types.FieldType_Decimal || param != scale {
			t.Errorf("Decimal(%d) 往返后变成 (%v,%d)", scale, ft, param)
		}
	}
}

// TestCompositePKNote 复合主键的报告必须点出"源表有主键, 只是多列"。
//
// 以前读端直接把主键丢成空串, 下游报的是"源表没有主键"—— 用户会去源库找一个不存在的
// 问题 (源表的主键好好的), 真正的原因是 VRTX 的行存储索引只能单列。
func TestCompositePKNote(t *testing.T) {
	note := compositePKNote("orders", []string{"tenant_id", "order_no"})
	for _, want := range []string{"orders", "tenant_id", "order_no", "复合键"} {
		if !strings.Contains(note, want) {
			t.Errorf("复合主键的报告应包含 %q: %s", want, note)
		}
	}
}

// ============================ 端到端: 闸门真的接在迁移路径上 ============================
//
// 上面测的是校验函数本身。一个"从来没被调用过"的校验函数同样能过那些用例, 所以这里从
// Run 走一遍: 源库 (SQLite, 与 MySQL 在"结构比 VRTX 宽"这点上同类) 真有这样的表时,
// 迁移是不是按预期拒绝 / 跳过。

// TestWideTableRejectedBeforeBuild 源库列数超过 VRTX 上限 → 建表之前就报错, 不留半成品。
//
// MySQL (InnoDB) 允许 1017 列而 VRTX 只到 core.MaxFieldsPerTable。不前置检查, 这条路会
// 一直走到 core 写 FieldCount 时才炸, 报出一句与源库无关的"字段数量非法"—— 用户不知道
// 该改源表还是该拆表。
func TestWideTableRejectedBeforeBuild(t *testing.T) {
	dir := t.TempDir()
	srcPath := filepath.Join(dir, "src.sqlite")
	dstPath := filepath.Join(dir, "dst.db")

	// 比上限多 32 列: 源库(SQLite)建得出来, VRTX 建不出来
	const n = core.MaxFieldsPerTable + 32
	cols := make([]string, 0, n)
	for i := 0; i < n; i++ {
		cols = append(cols, "c"+strconv.Itoa(i)+" INTEGER")
	}
	db, err := sql.Open("sqlite", srcPath)
	if err != nil {
		t.Fatal(err)
	}
	for _, s := range []string{
		`CREATE TABLE Wide (id INTEGER PRIMARY KEY, ` + strings.Join(cols, ", ") + `)`,
		`INSERT INTO Wide (id, c0) VALUES (1, 7)`,
	} {
		if _, err := db.Exec(s); err != nil {
			t.Fatalf("建源表: %v", err)
		}
	}
	_ = db.Close()

	s1, err := OpenSQLite(srcPath, quiet())
	if err != nil {
		t.Fatal(err)
	}
	defer s1.Close()
	v1, err := OpenVRTX(dstPath, quiet())
	if err != nil {
		t.Fatal(err)
	}
	defer v1.Close()

	_, err = Run(s1, v1, quiet())
	if err == nil {
		t.Fatal("列数超过 VRTX 上限的表必须被拒")
	}
	for _, want := range []string{
		"Wide",
		strconv.Itoa(n + 1),                  // 源表实际有多少列 (n 个 + 主键 id)
		strconv.Itoa(core.MaxFieldsPerTable), // VRTX 上限
	} {
		if !strings.Contains(err.Error(), want) {
			t.Errorf("列数超限的报错应包含 %q: %v", want, err)
		}
	}

	// 闸门在【建表之前】: 目标库里不该留下一张半成品 Wide
	names, err := v1.Tables()
	if err != nil {
		t.Fatal(err)
	}
	for _, nm := range names {
		if nm == "Wide" {
			t.Error("被拒的表不该在目标库里留下半成品")
		}
	}
}

// TestLongIndexNameSkippedNotFatal 索引名超长只该跳那一条索引, 不该让整张表陪葬。
//
// SQLite 的索引名几乎没有长度限制, VRTX 只到 core.IndexNameSize(64)。这类"源库合法、
// 目标装不下"的索引以前会直接 return err 把整张表拦下来 —— 数据本来一条都不丢, 却因为
// 一个名字迁不过去。跳过的同时必须写进报告 (静默丢索引等于悄悄改了结构)。
func TestLongIndexNameSkippedNotFatal(t *testing.T) {
	dir := t.TempDir()
	srcPath := filepath.Join(dir, "src.sqlite")
	dstPath := filepath.Join(dir, "dst.db")

	longIdx := "idx_" + strings.Repeat("x", core.IndexNameSize) // 比上限长
	db, err := sql.Open("sqlite", srcPath)
	if err != nil {
		t.Fatal(err)
	}
	for _, s := range []string{
		`CREATE TABLE T (id INTEGER PRIMARY KEY, name VARCHAR(32) NOT NULL)`,
		`CREATE INDEX ` + longIdx + ` ON T(name)`,
		`INSERT INTO T (id, name) VALUES (1, 'a')`,
	} {
		if _, err := db.Exec(s); err != nil {
			t.Fatalf("执行 %s: %v", s, err)
		}
	}
	_ = db.Close()

	s1, err := OpenSQLite(srcPath, quiet())
	if err != nil {
		t.Fatal(err)
	}
	defer s1.Close()
	v1, err := OpenVRTX(dstPath, quiet())
	if err != nil {
		t.Fatal(err)
	}
	defer v1.Close()

	rep, err := Run(s1, v1, quiet())
	if err != nil {
		t.Fatalf("一条索引建不出来不该让整张表失败: %v", err)
	}
	if rep.Rows != 1 {
		t.Errorf("数据应当照常迁过去, 实际 %d 行", rep.Rows)
	}
	if len(rep.Tables) != 1 {
		t.Fatalf("应当只有一张表的报告, 实际 %d", len(rep.Tables))
	}
	note := rep.Tables[0].Note
	if !strings.Contains(note, "跳过索引") {
		t.Errorf("跳过的索引必须报出来: %q", note)
	}
	// 报错要能让用户判断该怎么办: 得说清是"名字超过 64 字节"
	if !strings.Contains(note, strconv.Itoa(core.IndexNameSize)) {
		t.Errorf("跳过的索引报告应给出上限 %d: %q", core.IndexNameSize, note)
	}
}