TsDB2/migrate/sqltype_test.go

package migrate

import (
	"TsDB2/DB/core/types"
	"testing"
)

// ============================ SQL 类型 ↔ VRTX 类型 ============================
//
// 这一层是迁移的"翻译表", 出错的后果不是报错而是【静默改变列的语义】:
//   - BLOB 认成 TEXT → 二进制内容进了文本列;
//   - FLOAT 在 SQLite/MySQL 上按同一档收 → 单精度列被当成双精度 (或反过来);
//   - DECIMAL(18,2) 只读类型名不读 (p,s) → 小数位被抹平, 金额变成整数。
//
// 所以这里既钉住逐条映射, 也钉住"声明出去再读回来还是同一个类型"这个往返性质。

func TestInferSQLType(t *testing.T) {
	cases := []struct {
		decl  string
		mysql bool
		ft    types.FieldType
		param byte
	}{
		// 整数: 有符号 / 无符号分档, SQLite 的 INTEGER 一律按 8 字节收
		{"TINYINT", true, types.FieldType_Int8, 0},
		{"SMALLINT", true, types.FieldType_Int16, 0},
		{"INT", true, types.FieldType_Int32, 0},
		{"MEDIUMINT", true, types.FieldType_Int32, 0},
		{"BIGINT", true, types.FieldType_Int64, 0},
		{"INTEGER", false, types.FieldType_Int64, 0},
		{"TINYINT UNSIGNED", true, types.FieldType_1byte, 0},
		{"SMALLINT UNSIGNED", true, types.FieldType_2byte, 0},
		{"INT UNSIGNED", true, types.FieldType_4byte, 0},
		{"BIGINT UNSIGNED", true, types.FieldType_8byte, 0},

		// 浮点: MySQL 的 FLOAT 就是 4 字节, SQLite 只有 REAL 亲和性 (8 字节)
		{"FLOAT", true, types.FieldType_Float32, 0},
		{"FLOAT", false, types.FieldType_Float64, 0},
		{"DOUBLE", true, types.FieldType_Float64, 0},
		{"REAL", false, types.FieldType_Float64, 0},

		// 定点: (p,s) 必须读进来 —— 只读类型名会把小数位抹平
		{"DECIMAL(18,2)", true, types.FieldType_Decimal, 2},
		{"DECIMAL(10,0)", true, types.FieldType_Decimal, 0},
		{"NUMERIC(20,6)", false, types.FieldType_Decimal, 6},
		// 超出 128 位系数容量 → 退回 Float64 (与改动前一致), 不假装精确
		{"DECIMAL(65,30)", true, types.FieldType_Float64, 0},

		// 其余新类型
		{"BOOLEAN", false, types.FieldType_Bool, 0},
		{"BOOL", true, types.FieldType_Bool, 0},
		{"YEAR", true, types.FieldType_Year, 0},
		{"UUID", true, types.FieldType_UUID, 0},

		// 二进制大对象 vs 文本: 这是本次新增的区分
		{"BLOB", false, types.FieldType_Blob, 0},
		{"TINYBLOB", true, types.FieldType_Blob, 0},
		{"MEDIUMBLOB", true, types.FieldType_Blob, 0},
		{"LONGBLOB", true, types.FieldType_Blob, 0},
		{"VARBINARY(64)", true, types.FieldType_Blob, 0},
		{"TEXT", false, types.FieldType_Dynamicbyte, 0},
		{"LONGTEXT", true, types.FieldType_Dynamicbyte, 0},

		// 定长串按长度就近取档
		{"VARCHAR(8)", true, types.FieldType_16byte, 0},
		{"VARCHAR(200)", true, types.FieldType_256byte, 0},
		{"CHAR(32)", true, types.FieldType_32byte, 0},

		// 时间列落 8byte (Unix 纳秒), 不是动态列
		{"DATETIME", true, types.FieldType_8byte, 0},
		{"DATE", false, types.FieldType_8byte, 0},
	}
	for _, c := range cases {
		ft, param := inferSQLType(c.decl, c.mysql)
		if ft != c.ft || param != c.param {
			t.Errorf("inferSQLType(%q, mysql=%v) = (%v, %d), 期望 (%v, %d)",
				c.decl, c.mysql, ft, param, c.ft, c.param)
		}
	}
}

// TestMySQLColumnDecl 钉住"字符列的容量按字节算"这条规则 (information_schema 里
// CHARACTER_OCTET_LENGTH 那一列)。
//
// 这条是真实踩出来的: resource_info.resAlias 是 varchar(50) utf8mb4, 库里存了 73 字节
// 的中文别名。老代码拿 50 (字符数) 去取档得到 64byte, 整表迁移失败 —— 而 50 个字符在
// utf8mb4 下最多是 200 字节, 只有拿 200 去取档才装得下。
func TestMySQLColumnDecl(t *testing.T) {
	cases := []struct {
		name     string
		dataType string
		// information_schema 给的三组长度: 字节容量 / 精度 / 小数位
		charBytes, numPrec, numScale int64
		want                         types.FieldType
		wantParam                    byte
	}{
		// varchar(50) utf8mb4 → octet 200 → 256byte (老算法 50 → 64byte, 装不下中文)
		{"utf8mb4 varchar(50)", "varchar", 200, 0, 0, types.FieldType_256byte, 0},
		// varchar(12) utf8mb4 → octet 48 → 64byte
		{"utf8mb4 varchar(12)", "varchar", 48, 0, 0, types.FieldType_64byte, 0},
		// 单字节字符集不背 4 倍: latin1 的 varchar(50) 就是 50 字节
		{"latin1 varchar(50)", "varchar", 50, 0, 0, types.FieldType_64byte, 0},
		// 正好卡在档位边界上: varchar(64) utf8mb4 = 256 字节, 用满最后一档
		{"utf8mb4 varchar(64)", "varchar", 256, 0, 0, types.FieldType_256byte, 0},
		// 再多一个字符就超出所有定长档: 宁可变长也不开窄列 (变长只是没索引, 不会丢数据)
		{"utf8mb4 varchar(65)", "varchar", 260, 0, 0, types.FieldType_Dynamicbyte, 0},
		{"utf8mb4 char(1)", "char", 4, 0, 0, types.FieldType_16byte, 0},
		{"utf8mb4 text", "text", 65535, 0, 0, types.FieldType_Dynamicbyte, 0},
		{"blob", "blob", 65535, 0, 0, types.FieldType_Blob, 0},
		// 定点: (p,s) 要读进来, 小数位不能被抹平
		{"decimal(18,2)", "decimal", 20, 18, 2, types.FieldType_Decimal, 2},
		// 长度信息为 0 (数值/时间列没有 octet length) 时退回纯类型名
		{"int", "int", 0, 10, 0, types.FieldType_Int32, 0},
		{"datetime", "datetime", 0, 0, 0, types.FieldType_8byte, 0},
	}
	for _, c := range cases {
		decl := mysqlColumnDecl(c.dataType, c.charBytes, c.numPrec, c.numScale)
		ft, param := inferSQLType(decl, true)
		if ft != c.want || param != c.wantParam {
			t.Errorf("%s: %q → (%v, %d), 期望 (%v, %d)",
				c.name, decl, ft, param, c.want, c.wantParam)
		}
	}
}

// TestSQLTypeRoundTrip 声明出去的类型名必须能被自己读回同一个类型。
//
// 例外的几种都【不依赖 meta 就得靠宽度/语义兜底】, 这里显式列出来当文档:
//   - 窄整数在 SQLite 只有 INTEGER 一种写法, 读回一律 Int64;
//   - Decimal 在 SQLite 没有原生类型 (落 TEXT), 读回是 Dynamicbyte;
//   - UUID 在 MySQL 用 BINARY(16) 存, 读回被认成 Blob;
//   - Blob 与 Dynamicbyte 盘上同为动态列, 靠 meta 才能区分到具体哪一种。
//
// 也就是说: 需要"精确类型"的往返都依赖 __vrtx_meta (默认开启); 没有 meta 时
// 至少保证["数据不丢、语义不离谱"]。
func TestSQLTypeRoundTrip(t *testing.T) {
	type rt struct {
		ft    types.FieldType
		param byte
	}
	for _, mysql := range []bool{true, false} {
		// 这些类型在两个方言下都能原样往返
		exact := []rt{
			{types.FieldType_Blob, 0},
			{types.FieldType_Bool, 0},
			{types.FieldType_Year, 0},
			{types.FieldType_Float64, 0},
			{types.FieldType_Int64, 0},
		}
		if mysql {
			exact = append(exact,
				rt{types.FieldType_Int8, 0},
				rt{types.FieldType_Int16, 0},
				rt{types.FieldType_Int32, 0},
				rt{types.FieldType_1byte, 0},
				rt{types.FieldType_2byte, 0},
				rt{types.FieldType_4byte, 0},
				rt{types.FieldType_8byte, 0},
				rt{types.FieldType_Float32, 0},
				rt{types.FieldType_Decimal, 2},
			)
		}
		// 定长串按长度取档, 档位本身就是 16/32/64/128/256, 所以能精确往返
		for _, ft := range []types.FieldType{
			types.FieldType_16byte, types.FieldType_32byte,
			types.FieldType_64byte, types.FieldType_128byte, types.FieldType_256byte,
		} {
			exact = append(exact, rt{ft, 0})
		}

		for _, e := range exact {
			decl := sqlTypeName(e.ft, e.param, mysql)
			ft, param := inferSQLType(decl, mysql)
			if ft != e.ft || param != e.param {
				t.Errorf("mysql=%v: %v → %q → (%v, %d), 期望 (%v, %d)",
					mysql, e.ft, decl, ft, param, e.ft, e.param)
			}
		}
	}
}